Пишем готовый проект на Python: асинхронный сервис отслеживания цены акций - UPROGER

Готовый портфельный проект на Python: асинхронный сервис «Stock Guardian»
Отслеживает цены акций в реальном времени, рассылает push-уведомления при достижении триггеров и предоставляет REST + WebSocket API для фронтенда.

Почему проект ценен для портфолио

Показывает владение современным стеком: FastAPI + WebSockets + asyncio, PostgreSQL, Redis, Docker, CI/CD.
Демонстрирует продвинутые практики: типы (pydantic, mypy), тесты (pytest, pytest-asyncio), линтеры, GitHub Actions.
Подходит для live-демо: легко задеплоить на Render/Fly.io/Hetzner и показать работу в браузере.

Архитектура

┌──────────┐      WebSocket    ┌────────┐
│  Client  │◀──────────────────│ FastAPI│
└──────────┘                  └────┬───┘
          REST /alerts             │
                                    ▼
                         ┌────────────────────┐
                         │ Redis (Pub/Sub)    │
                         └──────┬─────────────┘
                                ▼
                    ┌────────────────────┐
                    │ worker (Celery)    │
                    │  ▸ fetch quotes    │
                    │  ▸ evaluate rules  │
                    └──────┬─────────────┘
                           ▼
                    ┌──────────────┐
                    │ PostgreSQL   │
                    └──────────────┘

Технологический стек

Компонент	Причина выбора
FastAPI	высокопроизводительный ASGI-фреймворк
WebSockets	мгновенные пуши без long-polling
PostgreSQL	надёжное хранилище сигналов и пользователей
Redis Pub/Sub	быстрая передача сообщений между worker’ами и API
Celery + Beat	фоновый парсинг котировок и планировщик
Docker Compose	единое окружение для dev/prod
GitHub Actions	автоматическое тестирование и линтинг
Pydantic/Mypy	статическая валидация схем и типов

Минимальная реализация

Создай репозиторий и структуру

stock_guardian/
 ├ app/
 │   ├ api/            # FastAPI endpoints
 │   ├ services/       # бизнес-логика
 │   ├ workers/        # Celery tasks
 │   └ core/           # конфигурация, модели
 ├ tests/
 ├ Dockerfile
 ├ docker-compose.yml
 └ pyproject.toml

Модель сигнала

# app/core/models.py
from pydantic import BaseModel, Field
from uuid import UUID, uuid4

class Alert(BaseModel):
    id: UUID = Field(default_factory=uuid4)
    symbol: str
    target_price: float
    direction: str  # "above" | "below"
    user_id: UUID

WebSocket-подписка

# app/api/ws.py
from fastapi import APIRouter, WebSocket, WebSocketDisconnect
from redis.asyncio import PubSub

router = APIRouter()

@router.websocket("/ws/{user_id}")
async def alerts_ws(ws: WebSocket, user_id: str):
    await ws.accept()
    pubsub: PubSub = request.app.state.redis.pubsub()
    await pubsub.subscribe(user_id)
    try:
        async for msg in pubsub.listen():
            if msg["type"] == "message":
                await ws.send_text(msg["data"].decode())
    except WebSocketDisconnect:
        await pubsub.unsubscribe(user_id)

Фоновый воркер

# app/workers/quotes.py
@app.task
def check_alerts():
    quotes = fetch_batch_prices()          # вызов API брокера
    for alert in load_active_alerts():
        price = quotes.get(alert.symbol)
        hit = price >= alert.target_price if alert.direction == "above" else price <= alert.target_price
        if hit:
            redis.publish(alert.user_id, json.dumps({"symbol": alert.symbol, "price": price}))
            deactivate(alert.id)

Тестирование

async def test_create_alert(async_client):
    resp = await async_client.post("/alerts", json={"symbol": "NVDA", "target_price": 1000, "direction": "above"})
    assert resp.status_code == 201

CI Workflow (.github/workflows/ci.yml)

name: CI
on: [push, pull_request]

jobs:
  test:
    runs-on: ubuntu-latest
    services:
      postgres:
        image: postgres:16
        env:
          POSTGRES_USER: test
          POSTGRES_PASSWORD: test
          POSTGRES_DB: stock
        ports: [ "5432:5432" ]
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with: {python-version: "3.12"}
      - run: pip install -r requirements.txt
      - run: pytest -q

Идеи для расширения