Initial project structure: MarketScanner - Fear-to-Fortune Trading Intelligence

Features:
- FastAPI backend with stocks, news, signals, watchlist, analytics endpoints
- React frontend with TailwindCSS dark mode trading dashboard
- Celery workers for news fetching, sentiment analysis, pattern detection
- TimescaleDB schema for time-series stock data
- Docker Compose setup for all services
- OpenAI integration for sentiment analysis

backend/Dockerfile

@@ -0,0 +1,26 @@
+FROM python:3.12-slim
+
+WORKDIR /app
+
+# Install system dependencies
+RUN apt-get update && apt-get install -y \
+    gcc \
+    libpq-dev \
+    curl \
+    && rm -rf /var/lib/apt/lists/*
+
+# Install Python dependencies
+COPY requirements.txt .
+RUN pip install --no-cache-dir -r requirements.txt
+
+# Copy application code
+COPY . .
+
+# Create logs directory
+RUN mkdir -p /app/logs
+
+# Expose port
+EXPOSE 8000
+
+# Run the application
+CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000", "--reload"]

backend/app/api/__init__.py

@@ -0,0 +1,16 @@
+"""
+API Router - Main entry point for all API routes
+"""
+
+from fastapi import APIRouter
+
+from app.api.endpoints import stocks, news, signals, watchlist, analytics
+
+router = APIRouter()
+
+# Include all endpoint routers
+router.include_router(stocks.router, prefix="/stocks", tags=["Stocks"])
+router.include_router(news.router, prefix="/news", tags=["News"])
+router.include_router(signals.router, prefix="/signals", tags=["Buy Signals"])
+router.include_router(watchlist.router, prefix="/watchlist", tags=["Watchlist"])
+router.include_router(analytics.router, prefix="/analytics", tags=["Analytics"])

backend/app/api/endpoints/__init__.py

@@ -0,0 +1 @@
+"""API endpoints module."""

backend/app/api/endpoints/analytics.py

@@ -0,0 +1,140 @@
+"""
+Analytics API Endpoints
+"""
+
+from typing import Optional
+from datetime import datetime, timedelta
+from fastapi import APIRouter, Depends, Query
+from sqlalchemy.ext.asyncio import AsyncSession
+from sqlalchemy import select, func
+
+from app.core.database import get_db
+from app.models.stock import Stock
+from app.models.news import NewsArticle
+from app.models.signal import BuySignal
+from app.models.panic import PanicEvent
+
+router = APIRouter()
+
+
+@router.get("/dashboard")
+async def get_dashboard_stats(db: AsyncSession = Depends(get_db)):
+    """Get overview statistics for the dashboard."""
+    now = datetime.utcnow()
+    last_24h = now - timedelta(hours=24)
+    last_7d = now - timedelta(days=7)
+    
+    # Count stocks
+    stocks_count = await db.execute(
+        select(func.count(Stock.id)).where(Stock.is_active == True)
+    )
+    
+    # Count news (last 24h)
+    news_count = await db.execute(
+        select(func.count(NewsArticle.id))
+        .where(NewsArticle.published_at >= last_24h)
+    )
+    
+    # Count active signals
+    signals_count = await db.execute(
+        select(func.count(BuySignal.id))
+        .where(BuySignal.status == "active")
+    )
+    
+    # Average sentiment (last 24h)
+    avg_sentiment = await db.execute(
+        select(func.avg(NewsArticle.sentiment_score))
+        .where(NewsArticle.published_at >= last_24h)
+        .where(NewsArticle.sentiment_score.isnot(None))
+    )
+    
+    return {
+        "stocks_tracked": stocks_count.scalar() or 0,
+        "news_last_24h": news_count.scalar() or 0,
+        "active_signals": signals_count.scalar() or 0,
+        "avg_sentiment_24h": round(avg_sentiment.scalar() or 0, 2),
+        "timestamp": now.isoformat(),
+    }
+
+
+@router.get("/sentiment/trend")
+async def get_sentiment_trend(
+    db: AsyncSession = Depends(get_db),
+    days: int = Query(7, ge=1, le=30, description="Number of days"),
+):
+    """Get sentiment trend over time."""
+    # This would aggregate sentiment by day
+    # Placeholder for now
+    return {
+        "days": days,
+        "trend": [],
+        "message": "Sentiment trend data will be populated by the workers",
+    }
+
+
+@router.get("/sector/panic")
+async def get_sector_panic_levels(db: AsyncSession = Depends(get_db)):
+    """Get current panic levels by sector."""
+    # This would calculate average sentiment by sector
+    # Placeholder for now
+    return {
+        "sectors": [],
+        "message": "Sector panic data will be populated by the workers",
+    }
+
+
+@router.get("/patterns/top")
+async def get_top_patterns(
+    db: AsyncSession = Depends(get_db),
+    limit: int = Query(10, ge=1, le=50),
+):
+    """Get top historical patterns with best recovery rates."""
+    # This would query historical_patterns table
+    # Placeholder for now
+    return {
+        "patterns": [],
+        "message": "Pattern data will be populated by the pattern matcher",
+    }
+
+
+@router.get("/panic-events/recent")
+async def get_recent_panic_events(
+    db: AsyncSession = Depends(get_db),
+    days: int = Query(30, ge=1, le=90, description="Number of days"),
+    limit: int = Query(20, ge=1, le=50),
+):
+    """Get recent panic events."""
+    since = datetime.utcnow() - timedelta(days=days)
+    
+    query = (
+        select(PanicEvent)
+        .where(PanicEvent.start_time >= since)
+        .order_by(PanicEvent.start_time.desc())
+        .limit(limit)
+    )
+    result = await db.execute(query)
+    events = result.scalars().all()
+    
+    return {
+        "days": days,
+        "count": len(events),
+        "events": events,
+    }
+
+
+@router.get("/performance")
+async def get_signal_performance(
+    db: AsyncSession = Depends(get_db),
+    days: int = Query(90, ge=30, le=365, description="Number of days to analyze"),
+):
+    """Get performance metrics for past signals."""
+    # This would analyze triggered signals and their outcomes
+    # Placeholder for now
+    return {
+        "days": days,
+        "total_signals": 0,
+        "triggered_signals": 0,
+        "avg_return": 0,
+        "win_rate": 0,
+        "message": "Performance data requires historical signal outcomes",
+    }

backend/app/api/endpoints/news.py

@@ -0,0 +1,122 @@
+"""
+News API Endpoints
+"""
+
+from typing import List, Optional
+from datetime import datetime, timedelta
+from uuid import UUID
+from fastapi import APIRouter, Depends, HTTPException, Query
+from sqlalchemy.ext.asyncio import AsyncSession
+from sqlalchemy import select, desc
+
+from app.core.database import get_db
+from app.models.news import NewsArticle
+from app.schemas.news import NewsResponse, NewsWithSentiment
+
+router = APIRouter()
+
+
+@router.get("/", response_model=List[NewsResponse])
+async def list_news(
+    db: AsyncSession = Depends(get_db),
+    source: Optional[str] = Query(None, description="Filter by source"),
+    sentiment: Optional[str] = Query(None, description="Filter by sentiment: positive, negative, neutral"),
+    hours: int = Query(24, ge=1, le=168, description="News from last N hours"),
+    skip: int = Query(0, ge=0),
+    limit: int = Query(50, ge=1, le=100),
+):
+    """List recent news articles."""
+    since = datetime.utcnow() - timedelta(hours=hours)
+    
+    query = select(NewsArticle).where(NewsArticle.published_at >= since)
+    
+    if source:
+        query = query.where(NewsArticle.source == source)
+    if sentiment:
+        query = query.where(NewsArticle.sentiment_label == sentiment)
+    
+    query = query.order_by(desc(NewsArticle.published_at)).offset(skip).limit(limit)
+    result = await db.execute(query)
+    return result.scalars().all()
+
+
+@router.get("/sources")
+async def list_sources(db: AsyncSession = Depends(get_db)):
+    """Get list of all news sources."""
+    query = select(NewsArticle.source).distinct()
+    result = await db.execute(query)
+    sources = [row[0] for row in result.fetchall() if row[0]]
+    return {"sources": sorted(sources)}
+
+
+@router.get("/stock/{symbol}", response_model=List[NewsWithSentiment])
+async def get_news_for_stock(
+    symbol: str,
+    db: AsyncSession = Depends(get_db),
+    hours: int = Query(72, ge=1, le=720, description="News from last N hours"),
+    skip: int = Query(0, ge=0),
+    limit: int = Query(50, ge=1, le=100),
+):
+    """Get news articles mentioning a specific stock."""
+    # This would use the news_stock_mentions join table
+    # For now, we search in title/content
+    since = datetime.utcnow() - timedelta(hours=hours)
+    search_term = f"%{symbol.upper()}%"
+    
+    query = (
+        select(NewsArticle)
+        .where(NewsArticle.published_at >= since)
+        .where(
+            (NewsArticle.title.ilike(search_term)) |
+            (NewsArticle.content.ilike(search_term))
+        )
+        .order_by(desc(NewsArticle.published_at))
+        .offset(skip)
+        .limit(limit)
+    )
+    result = await db.execute(query)
+    return result.scalars().all()
+
+
+@router.get("/panic")
+async def get_panic_news(
+    db: AsyncSession = Depends(get_db),
+    threshold: float = Query(-50.0, ge=-100, le=0, description="Sentiment threshold"),
+    hours: int = Query(24, ge=1, le=168, description="News from last N hours"),
+    limit: int = Query(20, ge=1, le=50),
+):
+    """Get the most panic-inducing news articles."""
+    since = datetime.utcnow() - timedelta(hours=hours)
+    
+    query = (
+        select(NewsArticle)
+        .where(NewsArticle.published_at >= since)
+        .where(NewsArticle.sentiment_score <= threshold)
+        .order_by(NewsArticle.sentiment_score.asc())
+        .limit(limit)
+    )
+    result = await db.execute(query)
+    articles = result.scalars().all()
+    
+    return {
+        "threshold": threshold,
+        "hours": hours,
+        "count": len(articles),
+        "articles": articles,
+    }
+
+
+@router.get("/{article_id}", response_model=NewsWithSentiment)
+async def get_article(
+    article_id: UUID,
+    db: AsyncSession = Depends(get_db),
+):
+    """Get a specific news article with full details."""
+    query = select(NewsArticle).where(NewsArticle.id == article_id)
+    result = await db.execute(query)
+    article = result.scalar_one_or_none()
+    
+    if not article:
+        raise HTTPException(status_code=404, detail="Article not found")
+    
+    return article

backend/app/api/endpoints/signals.py

@@ -0,0 +1,123 @@
+"""
+Buy Signals API Endpoints
+"""
+
+from typing import List, Optional
+from datetime import datetime, timedelta
+from uuid import UUID
+from fastapi import APIRouter, Depends, HTTPException, Query
+from sqlalchemy.ext.asyncio import AsyncSession
+from sqlalchemy import select, desc
+
+from app.core.database import get_db
+from app.models.signal import BuySignal
+from app.schemas.signal import SignalResponse, SignalWithDetails
+
+router = APIRouter()
+
+
+@router.get("/", response_model=List[SignalResponse])
+async def list_signals(
+    db: AsyncSession = Depends(get_db),
+    status: Optional[str] = Query("active", description="Signal status: active, triggered, expired"),
+    min_confidence: float = Query(0.5, ge=0, le=1, description="Minimum confidence score"),
+    skip: int = Query(0, ge=0),
+    limit: int = Query(50, ge=1, le=100),
+):
+    """List buy signals ordered by confidence."""
+    query = select(BuySignal).where(BuySignal.confidence_score >= min_confidence)
+    
+    if status:
+        query = query.where(BuySignal.status == status)
+    
+    query = query.order_by(desc(BuySignal.confidence_score)).offset(skip).limit(limit)
+    result = await db.execute(query)
+    return result.scalars().all()
+
+
+@router.get("/top")
+async def get_top_signals(
+    db: AsyncSession = Depends(get_db),
+    limit: int = Query(10, ge=1, le=20),
+):
+    """Get top buy signals by confidence score."""
+    query = (
+        select(BuySignal)
+        .where(BuySignal.status == "active")
+        .order_by(desc(BuySignal.confidence_score))
+        .limit(limit)
+    )
+    result = await db.execute(query)
+    signals = result.scalars().all()
+    
+    return {
+        "count": len(signals),
+        "signals": signals,
+    }
+
+
+@router.get("/stock/{symbol}", response_model=List[SignalResponse])
+async def get_signals_for_stock(
+    symbol: str,
+    db: AsyncSession = Depends(get_db),
+    include_historical: bool = Query(False, description="Include past signals"),
+):
+    """Get buy signals for a specific stock."""
+    # We need to join with stocks table
+    # For now, placeholder
+    return []
+
+
+@router.get("/{signal_id}", response_model=SignalWithDetails)
+async def get_signal(
+    signal_id: UUID,
+    db: AsyncSession = Depends(get_db),
+):
+    """Get detailed information about a specific signal."""
+    query = select(BuySignal).where(BuySignal.id == signal_id)
+    result = await db.execute(query)
+    signal = result.scalar_one_or_none()
+    
+    if not signal:
+        raise HTTPException(status_code=404, detail="Signal not found")
+    
+    return signal
+
+
+@router.post("/{signal_id}/trigger")
+async def trigger_signal(
+    signal_id: UUID,
+    db: AsyncSession = Depends(get_db),
+):
+    """Mark a signal as triggered (you bought the stock)."""
+    query = select(BuySignal).where(BuySignal.id == signal_id)
+    result = await db.execute(query)
+    signal = result.scalar_one_or_none()
+    
+    if not signal:
+        raise HTTPException(status_code=404, detail="Signal not found")
+    
+    signal.status = "triggered"
+    signal.triggered_at = datetime.utcnow()
+    await db.commit()
+    
+    return {"message": "Signal marked as triggered", "signal_id": signal_id}
+
+
+@router.post("/{signal_id}/dismiss")
+async def dismiss_signal(
+    signal_id: UUID,
+    db: AsyncSession = Depends(get_db),
+):
+    """Dismiss a signal (not interested)."""
+    query = select(BuySignal).where(BuySignal.id == signal_id)
+    result = await db.execute(query)
+    signal = result.scalar_one_or_none()
+    
+    if not signal:
+        raise HTTPException(status_code=404, detail="Signal not found")
+    
+    signal.status = "cancelled"
+    await db.commit()
+    
+    return {"message": "Signal dismissed", "signal_id": signal_id}

backend/app/api/endpoints/stocks.py

@@ -0,0 +1,131 @@
+"""
+Stocks API Endpoints
+"""
+
+from typing import List, Optional
+from uuid import UUID
+from fastapi import APIRouter, Depends, HTTPException, Query
+from sqlalchemy.ext.asyncio import AsyncSession
+from sqlalchemy import select, func
+
+from app.core.database import get_db
+from app.models.stock import Stock
+from app.schemas.stock import StockResponse, StockCreate, StockWithPrice
+
+router = APIRouter()
+
+
+@router.get("/", response_model=List[StockResponse])
+async def list_stocks(
+    db: AsyncSession = Depends(get_db),
+    sector: Optional[str] = Query(None, description="Filter by sector"),
+    industry: Optional[str] = Query(None, description="Filter by industry"),
+    search: Optional[str] = Query(None, description="Search by symbol or name"),
+    skip: int = Query(0, ge=0),
+    limit: int = Query(50, ge=1, le=100),
+):
+    """List all tracked stocks with optional filters."""
+    query = select(Stock).where(Stock.is_active == True)
+    
+    if sector:
+        query = query.where(Stock.sector == sector)
+    if industry:
+        query = query.where(Stock.industry == industry)
+    if search:
+        search_term = f"%{search}%"
+        query = query.where(
+            (Stock.symbol.ilike(search_term)) | 
+            (Stock.name.ilike(search_term))
+        )
+    
+    query = query.offset(skip).limit(limit).order_by(Stock.symbol)
+    result = await db.execute(query)
+    return result.scalars().all()
+
+
+@router.get("/sectors")
+async def list_sectors(db: AsyncSession = Depends(get_db)):
+    """Get list of all unique sectors."""
+    query = select(Stock.sector).distinct().where(Stock.is_active == True)
+    result = await db.execute(query)
+    sectors = [row[0] for row in result.fetchall() if row[0]]
+    return {"sectors": sorted(sectors)}
+
+
+@router.get("/industries")
+async def list_industries(
+    db: AsyncSession = Depends(get_db),
+    sector: Optional[str] = Query(None, description="Filter by sector"),
+):
+    """Get list of all unique industries."""
+    query = select(Stock.industry).distinct().where(Stock.is_active == True)
+    if sector:
+        query = query.where(Stock.sector == sector)
+    result = await db.execute(query)
+    industries = [row[0] for row in result.fetchall() if row[0]]
+    return {"industries": sorted(industries)}
+
+
+@router.get("/{symbol}", response_model=StockWithPrice)
+async def get_stock(
+    symbol: str,
+    db: AsyncSession = Depends(get_db),
+):
+    """Get detailed stock information including latest price."""
+    query = select(Stock).where(Stock.symbol == symbol.upper())
+    result = await db.execute(query)
+    stock = result.scalar_one_or_none()
+    
+    if not stock:
+        raise HTTPException(status_code=404, detail=f"Stock {symbol} not found")
+    
+    # TODO: Add latest price from stock_prices table
+    return stock
+
+
+@router.post("/", response_model=StockResponse)
+async def add_stock(
+    stock: StockCreate,
+    db: AsyncSession = Depends(get_db),
+):
+    """Add a new stock to track."""
+    # Check if already exists
+    existing = await db.execute(
+        select(Stock).where(Stock.symbol == stock.symbol.upper())
+    )
+    if existing.scalar_one_or_none():
+        raise HTTPException(
+            status_code=400, 
+            detail=f"Stock {stock.symbol} already exists"
+        )
+    
+    db_stock = Stock(
+        symbol=stock.symbol.upper(),
+        name=stock.name,
+        sector=stock.sector,
+        industry=stock.industry,
+        exchange=stock.exchange,
+        country=stock.country,
+    )
+    db.add(db_stock)
+    await db.commit()
+    await db.refresh(db_stock)
+    return db_stock
+
+
+@router.delete("/{symbol}")
+async def remove_stock(
+    symbol: str,
+    db: AsyncSession = Depends(get_db),
+):
+    """Remove a stock from tracking (soft delete)."""
+    query = select(Stock).where(Stock.symbol == symbol.upper())
+    result = await db.execute(query)
+    stock = result.scalar_one_or_none()
+    
+    if not stock:
+        raise HTTPException(status_code=404, detail=f"Stock {symbol} not found")
+    
+    stock.is_active = False
+    await db.commit()
+    return {"message": f"Stock {symbol} removed from tracking"}

backend/app/api/endpoints/watchlist.py

@@ -0,0 +1,141 @@
+"""
+Watchlist API Endpoints
+"""
+
+from typing import List, Optional
+from uuid import UUID
+from fastapi import APIRouter, Depends, HTTPException, Query
+from sqlalchemy.ext.asyncio import AsyncSession
+from sqlalchemy import select
+
+from app.core.database import get_db
+from app.models.watchlist import Watchlist
+from app.models.stock import Stock
+from app.schemas.watchlist import WatchlistResponse, WatchlistCreate, WatchlistUpdate
+
+router = APIRouter()
+
+
+@router.get("/", response_model=List[WatchlistResponse])
+async def list_watchlist(
+    db: AsyncSession = Depends(get_db),
+    priority: Optional[int] = Query(None, ge=1, le=3, description="Filter by priority"),
+):
+    """Get all items in watchlist."""
+    query = select(Watchlist).where(Watchlist.is_active == True)
+    
+    if priority:
+        query = query.where(Watchlist.priority == priority)
+    
+    query = query.order_by(Watchlist.priority)
+    result = await db.execute(query)
+    return result.scalars().all()
+
+
+@router.post("/", response_model=WatchlistResponse)
+async def add_to_watchlist(
+    item: WatchlistCreate,
+    db: AsyncSession = Depends(get_db),
+):
+    """Add a stock to the watchlist."""
+    # Find the stock
+    stock_query = select(Stock).where(Stock.symbol == item.symbol.upper())
+    stock_result = await db.execute(stock_query)
+    stock = stock_result.scalar_one_or_none()
+    
+    if not stock:
+        raise HTTPException(status_code=404, detail=f"Stock {item.symbol} not found")
+    
+    # Check if already in watchlist
+    existing = await db.execute(
+        select(Watchlist).where(Watchlist.stock_id == stock.id)
+    )
+    if existing.scalar_one_or_none():
+        raise HTTPException(
+            status_code=400,
+            detail=f"Stock {item.symbol} is already in watchlist"
+        )
+    
+    watchlist_item = Watchlist(
+        stock_id=stock.id,
+        panic_alert_threshold=item.panic_alert_threshold,
+        price_alert_low=item.price_alert_low,
+        price_alert_high=item.price_alert_high,
+        priority=item.priority,
+        notes=item.notes,
+    )
+    db.add(watchlist_item)
+    await db.commit()
+    await db.refresh(watchlist_item)
+    return watchlist_item
+
+
+@router.put("/{watchlist_id}", response_model=WatchlistResponse)
+async def update_watchlist_item(
+    watchlist_id: UUID,
+    update: WatchlistUpdate,
+    db: AsyncSession = Depends(get_db),
+):
+    """Update a watchlist item."""
+    query = select(Watchlist).where(Watchlist.id == watchlist_id)
+    result = await db.execute(query)
+    item = result.scalar_one_or_none()
+    
+    if not item:
+        raise HTTPException(status_code=404, detail="Watchlist item not found")
+    
+    update_data = update.model_dump(exclude_unset=True)
+    for key, value in update_data.items():
+        setattr(item, key, value)
+    
+    await db.commit()
+    await db.refresh(item)
+    return item
+
+
+@router.delete("/{watchlist_id}")
+async def remove_from_watchlist(
+    watchlist_id: UUID,
+    db: AsyncSession = Depends(get_db),
+):
+    """Remove a stock from the watchlist."""
+    query = select(Watchlist).where(Watchlist.id == watchlist_id)
+    result = await db.execute(query)
+    item = result.scalar_one_or_none()
+    
+    if not item:
+        raise HTTPException(status_code=404, detail="Watchlist item not found")
+    
+    await db.delete(item)
+    await db.commit()
+    return {"message": "Removed from watchlist"}
+
+
+@router.delete("/symbol/{symbol}")
+async def remove_symbol_from_watchlist(
+    symbol: str,
+    db: AsyncSession = Depends(get_db),
+):
+    """Remove a stock from watchlist by symbol."""
+    # Find the stock
+    stock_query = select(Stock).where(Stock.symbol == symbol.upper())
+    stock_result = await db.execute(stock_query)
+    stock = stock_result.scalar_one_or_none()
+    
+    if not stock:
+        raise HTTPException(status_code=404, detail=f"Stock {symbol} not found")
+    
+    # Find and remove watchlist item
+    watchlist_query = select(Watchlist).where(Watchlist.stock_id == stock.id)
+    watchlist_result = await db.execute(watchlist_query)
+    item = watchlist_result.scalar_one_or_none()
+    
+    if not item:
+        raise HTTPException(
+            status_code=404,
+            detail=f"Stock {symbol} is not in watchlist"
+        )
+    
+    await db.delete(item)
+    await db.commit()
+    return {"message": f"Removed {symbol} from watchlist"}

backend/app/core/__init__.py

@@ -0,0 +1,6 @@
+"""Core module exports."""
+
+from app.core.config import settings
+from app.core.database import get_db, Base, AsyncSessionLocal
+
+__all__ = ["settings", "get_db", "Base", "AsyncSessionLocal"]

backend/app/core/config.py

@@ -0,0 +1,97 @@
+"""
+Application Configuration
+"""
+
+from typing import List
+from pydantic_settings import BaseSettings
+from functools import lru_cache
+
+
+class Settings(BaseSettings):
+    """Application settings loaded from environment variables."""
+    
+    # Application
+    VERSION: str = "0.1.0"
+    DEBUG: bool = False
+    SECRET_KEY: str = "change-me-in-production"
+    
+    # Server
+    BACKEND_HOST: str = "0.0.0.0"
+    BACKEND_PORT: int = 8000
+    CORS_ORIGINS: List[str] = ["http://localhost:3000", "http://localhost:5173"]
+    
+    # Database
+    POSTGRES_HOST: str = "localhost"
+    POSTGRES_PORT: int = 5432
+    POSTGRES_DB: str = "marketscanner"
+    POSTGRES_USER: str = "marketscanner"
+    POSTGRES_PASSWORD: str = "changeme"
+    
+    @property
+    def DATABASE_URL(self) -> str:
+        return f"postgresql+asyncpg://{self.POSTGRES_USER}:{self.POSTGRES_PASSWORD}@{self.POSTGRES_HOST}:{self.POSTGRES_PORT}/{self.POSTGRES_DB}"
+    
+    @property
+    def DATABASE_URL_SYNC(self) -> str:
+        return f"postgresql://{self.POSTGRES_USER}:{self.POSTGRES_PASSWORD}@{self.POSTGRES_HOST}:{self.POSTGRES_PORT}/{self.POSTGRES_DB}"
+    
+    # Redis
+    REDIS_HOST: str = "localhost"
+    REDIS_PORT: int = 6379
+    REDIS_PASSWORD: str = ""
+    
+    @property
+    def REDIS_URL(self) -> str:
+        if self.REDIS_PASSWORD:
+            return f"redis://:{self.REDIS_PASSWORD}@{self.REDIS_HOST}:{self.REDIS_PORT}/0"
+        return f"redis://{self.REDIS_HOST}:{self.REDIS_PORT}/0"
+    
+    # RabbitMQ
+    RABBITMQ_HOST: str = "localhost"
+    RABBITMQ_PORT: int = 5672
+    RABBITMQ_USER: str = "guest"
+    RABBITMQ_PASSWORD: str = "guest"
+    
+    @property
+    def RABBITMQ_URL(self) -> str:
+        return f"amqp://{self.RABBITMQ_USER}:{self.RABBITMQ_PASSWORD}@{self.RABBITMQ_HOST}:{self.RABBITMQ_PORT}//"
+    
+    # API Keys - Stock Data
+    ALPHA_VANTAGE_API_KEY: str = ""
+    POLYGON_API_KEY: str = ""
+    YAHOO_FINANCE_ENABLED: bool = True
+    FINNHUB_API_KEY: str = ""
+    
+    # API Keys - News
+    NEWS_API_KEY: str = ""
+    
+    # API Keys - AI
+    OPENAI_API_KEY: str = ""
+    OPENAI_MODEL: str = "gpt-4o-mini"
+    USE_LOCAL_LLM: bool = False
+    OLLAMA_HOST: str = "http://localhost:11434"
+    OLLAMA_MODEL: str = "llama3.2"
+    
+    # Scanning Settings
+    NEWS_SCAN_INTERVAL: int = 300  # seconds
+    STOCK_PRICE_INTERVAL: int = 60  # seconds
+    MAX_TRACKED_STOCKS: int = 500
+    PANIC_THRESHOLD: float = -50.0
+    
+    # Alerts
+    TELEGRAM_BOT_TOKEN: str = ""
+    TELEGRAM_CHAT_ID: str = ""
+    DISCORD_WEBHOOK_URL: str = ""
+    
+    class Config:
+        env_file = ".env"
+        case_sensitive = True
+
+
+@lru_cache()
+def get_settings() -> Settings:
+    """Get cached settings instance."""
+    return Settings()
+
+
+settings = get_settings()

backend/app/core/database.py

@@ -0,0 +1,79 @@
+"""
+Database Configuration and Session Management
+"""
+
+from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine, async_sessionmaker
+from sqlalchemy.orm import declarative_base
+from sqlalchemy import create_engine
+from sqlalchemy.orm import sessionmaker
+
+from app.core.config import settings
+
+# Async engine for FastAPI
+async_engine = create_async_engine(
+    settings.DATABASE_URL,
+    echo=settings.DEBUG,
+    pool_pre_ping=True,
+    pool_size=10,
+    max_overflow=20,
+)
+
+# Async session factory
+AsyncSessionLocal = async_sessionmaker(
+    async_engine,
+    class_=AsyncSession,
+    expire_on_commit=False,
+    autocommit=False,
+    autoflush=False,
+)
+
+# Sync engine for Celery workers
+sync_engine = create_engine(
+    settings.DATABASE_URL_SYNC,
+    echo=settings.DEBUG,
+    pool_pre_ping=True,
+    pool_size=5,
+    max_overflow=10,
+)
+
+# Sync session factory
+SyncSessionLocal = sessionmaker(
+    sync_engine,
+    autocommit=False,
+    autoflush=False,
+)
+
+# Base class for models
+Base = declarative_base()
+
+
+async def init_db():
+    """Initialize database (create tables if needed)."""
+    # Tables are created by init.sql, but we can add migrations here
+    pass
+
+
+async def get_db() -> AsyncSession:
+    """Dependency for getting async database session."""
+    async with AsyncSessionLocal() as session:
+        try:
+            yield session
+            await session.commit()
+        except Exception:
+            await session.rollback()
+            raise
+        finally:
+            await session.close()
+
+
+def get_sync_db():
+    """Get sync database session for Celery workers."""
+    db = SyncSessionLocal()
+    try:
+        yield db
+        db.commit()
+    except Exception:
+        db.rollback()
+        raise
+    finally:
+        db.close()

backend/app/main.py

@@ -0,0 +1,114 @@
+"""
+MarketScanner - Fear-to-Fortune Trading Intelligence
+Main FastAPI Application
+"""
+
+from contextlib import asynccontextmanager
+from fastapi import FastAPI
+from fastapi.middleware.cors import CORSMiddleware
+from fastapi.responses import JSONResponse
+import structlog
+
+from app.core.config import settings
+from app.core.database import init_db
+from app.api import router as api_router
+
+# Configure structured logging
+structlog.configure(
+    processors=[
+        structlog.stdlib.filter_by_level,
+        structlog.processors.TimeStamper(fmt="iso"),
+        structlog.processors.JSONRenderer()
+    ],
+    wrapper_class=structlog.stdlib.BoundLogger,
+    context_class=dict,
+    logger_factory=structlog.stdlib.LoggerFactory(),
+)
+
+logger = structlog.get_logger()
+
+
+@asynccontextmanager
+async def lifespan(app: FastAPI):
+    """Application lifespan events."""
+    # Startup
+    logger.info("Starting MarketScanner API", version=settings.VERSION)
+    await init_db()
+    logger.info("Database initialized")
+    
+    yield
+    
+    # Shutdown
+    logger.info("Shutting down MarketScanner API")
+
+
+# Create FastAPI application
+app = FastAPI(
+    title="MarketScanner API",
+    description="""
+    🚀 **MarketScanner** - Fear-to-Fortune Trading Intelligence
+    
+    A system that identifies buying opportunities by analyzing how stocks 
+    historically respond to panic-inducing news.
+    
+    ## Features
+    
+    * **News Monitoring** - Real-time scanning of financial news
+    * **Sentiment Analysis** - NLP-powered sentiment scoring
+    * **Panic Detection** - Identify market fear events
+    * **Pattern Matching** - Historical recovery patterns
+    * **Buy Signals** - Confidence-scored opportunities
+    
+    *"Buy when there's blood in the streets."* — Baron Rothschild
+    """,
+    version=settings.VERSION,
+    docs_url="/docs",
+    redoc_url="/redoc",
+    lifespan=lifespan,
+)
+
+# CORS middleware
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=settings.CORS_ORIGINS,
+    allow_credentials=True,
+    allow_methods=["*"],
+    allow_headers=["*"],
+)
+
+# Include API routes
+app.include_router(api_router, prefix="/api/v1")
+
+
+@app.get("/", tags=["Root"])
+async def root():
+    """Root endpoint with API info."""
+    return {
+        "name": "MarketScanner API",
+        "version": settings.VERSION,
+        "description": "Fear-to-Fortune Trading Intelligence",
+        "docs": "/docs",
+        "health": "/health",
+    }
+
+
+@app.get("/health", tags=["Health"])
+async def health_check():
+    """Health check endpoint for Docker/Kubernetes."""
+    return JSONResponse(
+        status_code=200,
+        content={
+            "status": "healthy",
+            "version": settings.VERSION,
+        }
+    )
+
+
+if __name__ == "__main__":
+    import uvicorn
+    uvicorn.run(
+        "app.main:app",
+        host=settings.BACKEND_HOST,
+        port=settings.BACKEND_PORT,
+        reload=settings.DEBUG,
+    )

backend/app/models/__init__.py

@@ -0,0 +1,9 @@
+"""Database models."""
+
+from app.models.stock import Stock
+from app.models.news import NewsArticle
+from app.models.signal import BuySignal
+from app.models.panic import PanicEvent
+from app.models.watchlist import Watchlist
+
+__all__ = ["Stock", "NewsArticle", "BuySignal", "PanicEvent", "Watchlist"]

backend/app/models/news.py

@@ -0,0 +1,34 @@
+"""News article model."""
+
+from sqlalchemy import Column, String, Text, Boolean, DateTime, Numeric
+from sqlalchemy.dialects.postgresql import UUID
+from sqlalchemy.sql import func
+import uuid
+
+from app.core.database import Base
+
+
+class NewsArticle(Base):
+    """News article table model."""
+    
+    __tablename__ = "news_articles"
+    
+    id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
+    title = Column(Text, nullable=False)
+    content = Column(Text)
+    summary = Column(Text)
+    url = Column(Text, unique=True, nullable=False)
+    source = Column(String(100), nullable=False, index=True)
+    author = Column(String(255))
+    published_at = Column(DateTime(timezone=True), nullable=False, index=True)
+    fetched_at = Column(DateTime(timezone=True), server_default=func.now())
+    image_url = Column(Text)
+    
+    # Sentiment analysis results
+    sentiment_score = Column(Numeric(5, 2), index=True)  # -100 to +100
+    sentiment_label = Column(String(20))  # negative, neutral, positive
+    sentiment_confidence = Column(Numeric(5, 4))
+    
+    # Processing status
+    is_processed = Column(Boolean, default=False)
+    processing_error = Column(Text)

backend/app/models/panic.py

@@ -0,0 +1,48 @@
+"""Panic event model."""
+
+from sqlalchemy import Column, String, Text, Boolean, DateTime, Numeric, Integer, ForeignKey
+from sqlalchemy.dialects.postgresql import UUID
+from sqlalchemy.sql import func
+import uuid
+
+from app.core.database import Base
+
+
+class PanicEvent(Base):
+    """Panic event table model."""
+    
+    __tablename__ = "panic_events"
+    
+    id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
+    stock_id = Column(UUID(as_uuid=True), ForeignKey("stocks.id", ondelete="CASCADE"), nullable=False, index=True)
+    
+    # Event timing
+    start_time = Column(DateTime(timezone=True), nullable=False, index=True)
+    peak_time = Column(DateTime(timezone=True))
+    end_time = Column(DateTime(timezone=True))
+    
+    # Price impact
+    price_at_start = Column(Numeric(15, 4), nullable=False)
+    price_at_peak_panic = Column(Numeric(15, 4))
+    price_at_end = Column(Numeric(15, 4))
+    max_drawdown_percent = Column(Numeric(8, 4), index=True)
+    
+    # Sentiment
+    avg_sentiment_score = Column(Numeric(5, 2))
+    min_sentiment_score = Column(Numeric(5, 2))
+    news_volume = Column(Integer)
+    
+    # Recovery metrics
+    recovery_time_days = Column(Integer)
+    recovery_percent = Column(Numeric(8, 4))
+    
+    # Classification
+    event_type = Column(String(100), index=True)  # earnings_miss, scandal, lawsuit, macro, etc.
+    event_category = Column(String(50))  # company_specific, sector_wide, market_wide
+    
+    # Analysis
+    is_complete = Column(Boolean, default=False)
+    notes = Column(Text)
+    
+    created_at = Column(DateTime(timezone=True), server_default=func.now())
+    updated_at = Column(DateTime(timezone=True), server_default=func.now(), onupdate=func.now())

backend/app/models/signal.py

@@ -0,0 +1,46 @@
+"""Buy signal model."""
+
+from sqlalchemy import Column, String, DateTime, Numeric, Integer, ForeignKey
+from sqlalchemy.dialects.postgresql import UUID
+from sqlalchemy.sql import func
+import uuid
+
+from app.core.database import Base
+
+
+class BuySignal(Base):
+    """Buy signal table model."""
+    
+    __tablename__ = "buy_signals"
+    
+    id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
+    stock_id = Column(UUID(as_uuid=True), ForeignKey("stocks.id", ondelete="CASCADE"), nullable=False, index=True)
+    panic_event_id = Column(UUID(as_uuid=True), ForeignKey("panic_events.id", ondelete="SET NULL"))
+    
+    # Signal details
+    signal_time = Column(DateTime(timezone=True), nullable=False, server_default=func.now(), index=True)
+    signal_price = Column(Numeric(15, 4), nullable=False)
+    
+    # Confidence scoring
+    confidence_score = Column(Numeric(5, 4), nullable=False, index=True)  # 0 to 1
+    
+    # Based on pattern
+    pattern_id = Column(UUID(as_uuid=True), ForeignKey("historical_patterns.id", ondelete="SET NULL"))
+    expected_recovery_percent = Column(Numeric(8, 4))
+    expected_recovery_days = Column(Integer)
+    
+    # Current metrics
+    current_drawdown_percent = Column(Numeric(8, 4))
+    current_sentiment_score = Column(Numeric(5, 2))
+    
+    # Signal status
+    status = Column(String(20), default="active", index=True)  # active, triggered, expired, cancelled
+    triggered_at = Column(DateTime(timezone=True))
+    
+    # Outcome tracking
+    outcome_price = Column(Numeric(15, 4))
+    outcome_percent = Column(Numeric(8, 4))
+    outcome_days = Column(Integer)
+    
+    created_at = Column(DateTime(timezone=True), server_default=func.now())
+    updated_at = Column(DateTime(timezone=True), server_default=func.now(), onupdate=func.now())

backend/app/models/stock.py

@@ -0,0 +1,26 @@
+"""Stock model."""
+
+from sqlalchemy import Column, String, BigInteger, Boolean, DateTime
+from sqlalchemy.dialects.postgresql import UUID
+from sqlalchemy.sql import func
+import uuid
+
+from app.core.database import Base
+
+
+class Stock(Base):
+    """Stock table model."""
+    
+    __tablename__ = "stocks"
+    
+    id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
+    symbol = Column(String(20), unique=True, nullable=False, index=True)
+    name = Column(String(255), nullable=False)
+    sector = Column(String(100), index=True)
+    industry = Column(String(100), index=True)
+    market_cap = Column(BigInteger)
+    exchange = Column(String(50))
+    country = Column(String(100), default="USA")
+    is_active = Column(Boolean, default=True)
+    created_at = Column(DateTime(timezone=True), server_default=func.now())
+    updated_at = Column(DateTime(timezone=True), server_default=func.now(), onupdate=func.now())

backend/app/models/watchlist.py

@@ -0,0 +1,30 @@
+"""Watchlist model."""
+
+from sqlalchemy import Column, String, Text, Boolean, DateTime, Numeric, Integer, ForeignKey
+from sqlalchemy.dialects.postgresql import UUID
+from sqlalchemy.sql import func
+import uuid
+
+from app.core.database import Base
+
+
+class Watchlist(Base):
+    """Watchlist table model."""
+    
+    __tablename__ = "watchlist"
+    
+    id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
+    stock_id = Column(UUID(as_uuid=True), ForeignKey("stocks.id", ondelete="CASCADE"), nullable=False, unique=True)
+    
+    # Alert thresholds
+    panic_alert_threshold = Column(Numeric(5, 2), default=-50)
+    price_alert_low = Column(Numeric(15, 4))
+    price_alert_high = Column(Numeric(15, 4))
+    
+    # Preferences
+    priority = Column(Integer, default=1, index=True)  # 1 = high, 2 = medium, 3 = low
+    notes = Column(Text)
+    
+    is_active = Column(Boolean, default=True)
+    created_at = Column(DateTime(timezone=True), server_default=func.now())
+    updated_at = Column(DateTime(timezone=True), server_default=func.now(), onupdate=func.now())

backend/app/schemas/__init__.py

@@ -0,0 +1,13 @@
+"""Pydantic schemas."""
+
+from app.schemas.stock import StockCreate, StockResponse, StockWithPrice
+from app.schemas.news import NewsCreate, NewsResponse, NewsWithSentiment
+from app.schemas.signal import SignalResponse, SignalWithDetails
+from app.schemas.watchlist import WatchlistCreate, WatchlistResponse, WatchlistUpdate
+
+__all__ = [
+    "StockCreate", "StockResponse", "StockWithPrice",
+    "NewsCreate", "NewsResponse", "NewsWithSentiment",
+    "SignalResponse", "SignalWithDetails",
+    "WatchlistCreate", "WatchlistResponse", "WatchlistUpdate",
+]

backend/app/schemas/news.py

@@ -0,0 +1,43 @@
+"""News schemas."""
+
+from typing import Optional
+from datetime import datetime
+from uuid import UUID
+from pydantic import BaseModel, Field
+
+
+class NewsBase(BaseModel):
+    """Base news schema."""
+    title: str
+    url: str
+    source: str
+    published_at: datetime
+
+
+class NewsCreate(NewsBase):
+    """Schema for creating a news article."""
+    content: Optional[str] = None
+    summary: Optional[str] = None
+    author: Optional[str] = None
+    image_url: Optional[str] = None
+
+
+class NewsResponse(NewsBase):
+    """Schema for news response."""
+    id: UUID
+    summary: Optional[str] = None
+    author: Optional[str] = None
+    fetched_at: datetime
+    
+    class Config:
+        from_attributes = True
+
+
+class NewsWithSentiment(NewsResponse):
+    """News response with sentiment analysis."""
+    content: Optional[str] = None
+    image_url: Optional[str] = None
+    sentiment_score: Optional[float] = None
+    sentiment_label: Optional[str] = None
+    sentiment_confidence: Optional[float] = None
+    is_processed: bool = False

backend/app/schemas/signal.py

@@ -0,0 +1,39 @@
+"""Signal schemas."""
+
+from typing import Optional
+from datetime import datetime
+from uuid import UUID
+from pydantic import BaseModel
+
+
+class SignalBase(BaseModel):
+    """Base signal schema."""
+    stock_id: UUID
+    signal_price: float
+    confidence_score: float
+
+
+class SignalResponse(SignalBase):
+    """Schema for signal response."""
+    id: UUID
+    signal_time: datetime
+    status: str
+    current_drawdown_percent: Optional[float] = None
+    current_sentiment_score: Optional[float] = None
+    expected_recovery_percent: Optional[float] = None
+    expected_recovery_days: Optional[int] = None
+    created_at: datetime
+    
+    class Config:
+        from_attributes = True
+
+
+class SignalWithDetails(SignalResponse):
+    """Signal response with full details."""
+    panic_event_id: Optional[UUID] = None
+    pattern_id: Optional[UUID] = None
+    triggered_at: Optional[datetime] = None
+    outcome_price: Optional[float] = None
+    outcome_percent: Optional[float] = None
+    outcome_days: Optional[int] = None
+    updated_at: datetime

backend/app/schemas/stock.py

@@ -0,0 +1,41 @@
+"""Stock schemas."""
+
+from typing import Optional
+from datetime import datetime
+from uuid import UUID
+from pydantic import BaseModel, Field
+
+
+class StockBase(BaseModel):
+    """Base stock schema."""
+    symbol: str = Field(..., min_length=1, max_length=20)
+    name: str = Field(..., min_length=1, max_length=255)
+    sector: Optional[str] = None
+    industry: Optional[str] = None
+    exchange: Optional[str] = None
+    country: str = "USA"
+
+
+class StockCreate(StockBase):
+    """Schema for creating a stock."""
+    pass
+
+
+class StockResponse(StockBase):
+    """Schema for stock response."""
+    id: UUID
+    market_cap: Optional[int] = None
+    is_active: bool
+    created_at: datetime
+    updated_at: datetime
+    
+    class Config:
+        from_attributes = True
+
+
+class StockWithPrice(StockResponse):
+    """Stock response with latest price data."""
+    latest_price: Optional[float] = None
+    price_change_24h: Optional[float] = None
+    price_change_percent_24h: Optional[float] = None
+    volume_24h: Optional[int] = None

backend/app/schemas/watchlist.py

@@ -0,0 +1,42 @@
+"""Watchlist schemas."""
+
+from typing import Optional
+from datetime import datetime
+from uuid import UUID
+from pydantic import BaseModel, Field
+
+
+class WatchlistBase(BaseModel):
+    """Base watchlist schema."""
+    panic_alert_threshold: float = -50.0
+    price_alert_low: Optional[float] = None
+    price_alert_high: Optional[float] = None
+    priority: int = Field(1, ge=1, le=3)
+    notes: Optional[str] = None
+
+
+class WatchlistCreate(WatchlistBase):
+    """Schema for creating a watchlist item."""
+    symbol: str
+
+
+class WatchlistUpdate(BaseModel):
+    """Schema for updating a watchlist item."""
+    panic_alert_threshold: Optional[float] = None
+    price_alert_low: Optional[float] = None
+    price_alert_high: Optional[float] = None
+    priority: Optional[int] = Field(None, ge=1, le=3)
+    notes: Optional[str] = None
+    is_active: Optional[bool] = None
+
+
+class WatchlistResponse(WatchlistBase):
+    """Schema for watchlist response."""
+    id: UUID
+    stock_id: UUID
+    is_active: bool
+    created_at: datetime
+    updated_at: datetime
+    
+    class Config:
+        from_attributes = True

backend/app/workers/__init__.py

@@ -0,0 +1,5 @@
+"""Celery workers module."""
+
+from app.workers.celery_app import celery_app
+
+__all__ = ["celery_app"]

backend/app/workers/celery_app.py

@@ -0,0 +1,69 @@
+"""
+Celery Application Configuration
+"""
+
+from celery import Celery
+from celery.schedules import crontab
+
+from app.core.config import settings
+
+# Create Celery app
+celery_app = Celery(
+    "marketscanner",
+    broker=settings.RABBITMQ_URL,
+    backend=settings.REDIS_URL,
+    include=[
+        "app.workers.tasks.news_tasks",
+        "app.workers.tasks.stock_tasks",
+        "app.workers.tasks.sentiment_tasks",
+        "app.workers.tasks.pattern_tasks",
+        "app.workers.tasks.alert_tasks",
+    ],
+)
+
+# Celery configuration
+celery_app.conf.update(
+    task_serializer="json",
+    accept_content=["json"],
+    result_serializer="json",
+    timezone="UTC",
+    enable_utc=True,
+    task_track_started=True,
+    task_time_limit=300,  # 5 minutes
+    worker_prefetch_multiplier=1,
+    worker_concurrency=4,
+)
+
+# Beat schedule (periodic tasks)
+celery_app.conf.beat_schedule = {
+    # Fetch news every 5 minutes
+    "fetch-news-every-5-minutes": {
+        "task": "app.workers.tasks.news_tasks.fetch_all_news",
+        "schedule": settings.NEWS_SCAN_INTERVAL,
+    },
+    # Update stock prices every minute
+    "update-prices-every-minute": {
+        "task": "app.workers.tasks.stock_tasks.update_stock_prices",
+        "schedule": settings.STOCK_PRICE_INTERVAL,
+    },
+    # Process unanalyzed news every 2 minutes
+    "analyze-sentiment-every-2-minutes": {
+        "task": "app.workers.tasks.sentiment_tasks.process_unanalyzed_news",
+        "schedule": 120,
+    },
+    # Detect panic events every 5 minutes
+    "detect-panic-every-5-minutes": {
+        "task": "app.workers.tasks.pattern_tasks.detect_panic_events",
+        "schedule": 300,
+    },
+    # Generate signals every 10 minutes
+    "generate-signals-every-10-minutes": {
+        "task": "app.workers.tasks.pattern_tasks.generate_buy_signals",
+        "schedule": 600,
+    },
+    # Clean old data daily at midnight
+    "cleanup-daily": {
+        "task": "app.workers.tasks.news_tasks.cleanup_old_news",
+        "schedule": crontab(hour=0, minute=0),
+    },
+}

backend/app/workers/tasks/__init__.py

@@ -0,0 +1 @@
+"""Worker tasks module."""

backend/app/workers/tasks/alert_tasks.py

@@ -0,0 +1,153 @@
+"""
+Alert notification tasks
+"""
+
+import httpx
+import structlog
+
+from app.workers.celery_app import celery_app
+from app.core.config import settings
+
+logger = structlog.get_logger()
+
+
+@celery_app.task(name="app.workers.tasks.alert_tasks.send_telegram_alert")
+def send_telegram_alert(message: str):
+    """Send alert via Telegram."""
+    if not settings.TELEGRAM_BOT_TOKEN or not settings.TELEGRAM_CHAT_ID:
+        logger.warning("Telegram not configured")
+        return {"sent": False, "reason": "not_configured"}
+    
+    try:
+        url = f"https://api.telegram.org/bot{settings.TELEGRAM_BOT_TOKEN}/sendMessage"
+        payload = {
+            "chat_id": settings.TELEGRAM_CHAT_ID,
+            "text": message,
+            "parse_mode": "Markdown",
+        }
+        
+        with httpx.Client() as client:
+            response = client.post(url, json=payload)
+            response.raise_for_status()
+        
+        logger.info("Telegram alert sent")
+        return {"sent": True}
+        
+    except Exception as e:
+        logger.error("Failed to send Telegram alert", error=str(e))
+        return {"sent": False, "error": str(e)}
+
+
+@celery_app.task(name="app.workers.tasks.alert_tasks.send_discord_alert")
+def send_discord_alert(message: str, embed: dict = None):
+    """Send alert via Discord webhook."""
+    if not settings.DISCORD_WEBHOOK_URL:
+        logger.warning("Discord not configured")
+        return {"sent": False, "reason": "not_configured"}
+    
+    try:
+        payload = {"content": message}
+        if embed:
+            payload["embeds"] = [embed]
+        
+        with httpx.Client() as client:
+            response = client.post(settings.DISCORD_WEBHOOK_URL, json=payload)
+            response.raise_for_status()
+        
+        logger.info("Discord alert sent")
+        return {"sent": True}
+        
+    except Exception as e:
+        logger.error("Failed to send Discord alert", error=str(e))
+        return {"sent": False, "error": str(e)}
+
+
+@celery_app.task(name="app.workers.tasks.alert_tasks.send_buy_signal_alert")
+def send_buy_signal_alert(signal_data: dict):
+    """Send formatted buy signal alert to all configured channels."""
+    logger.info("Sending buy signal alert", symbol=signal_data.get("symbol"))
+    
+    # Format message
+    symbol = signal_data.get("symbol", "UNKNOWN")
+    confidence = signal_data.get("confidence", 0) * 100
+    current_price = signal_data.get("price", 0)
+    drawdown = signal_data.get("drawdown", 0)
+    expected_recovery = signal_data.get("expected_recovery", 0)
+    
+    message = f"""
+🚨 *BUY SIGNAL: ${symbol}* 🚨
+
+📊 *Confidence:* {confidence:.1f}%
+💰 *Current Price:* ${current_price:.2f}
+📉 *Drawdown:* {drawdown:.1f}%
+📈 *Expected Recovery:* {expected_recovery:.1f}%
+
+_"Buy when there's blood in the streets"_
+    """.strip()
+    
+    results = {
+        "telegram": None,
+        "discord": None,
+    }
+    
+    # Send to Telegram
+    if settings.TELEGRAM_BOT_TOKEN:
+        results["telegram"] = send_telegram_alert.delay(message).get()
+    
+    # Send to Discord with embed
+    if settings.DISCORD_WEBHOOK_URL:
+        embed = {
+            "title": f"🚨 BUY SIGNAL: ${symbol}",
+            "color": 0x00ff00,  # Green
+            "fields": [
+                {"name": "Confidence", "value": f"{confidence:.1f}%", "inline": True},
+                {"name": "Price", "value": f"${current_price:.2f}", "inline": True},
+                {"name": "Drawdown", "value": f"{drawdown:.1f}%", "inline": True},
+                {"name": "Expected Recovery", "value": f"{expected_recovery:.1f}%", "inline": True},
+            ],
+            "footer": {"text": "MarketScanner • Buy the Fear"},
+        }
+        results["discord"] = send_discord_alert.delay("", embed).get()
+    
+    return results
+
+
+@celery_app.task(name="app.workers.tasks.alert_tasks.send_panic_alert")
+def send_panic_alert(panic_data: dict):
+    """Send formatted panic detection alert."""
+    logger.info("Sending panic alert", symbol=panic_data.get("symbol"))
+    
+    symbol = panic_data.get("symbol", "UNKNOWN")
+    sentiment = panic_data.get("sentiment", 0)
+    price_drop = panic_data.get("price_drop", 0)
+    news_count = panic_data.get("news_count", 0)
+    
+    message = f"""
+🔴 *PANIC DETECTED: ${symbol}* 🔴
+
+😱 *Sentiment Score:* {sentiment:.1f}
+📉 *Price Drop:* {price_drop:.1f}%
+📰 *News Volume:* {news_count} articles
+
+⏳ Monitoring for buying opportunity...
+    """.strip()
+    
+    results = {}
+    
+    if settings.TELEGRAM_BOT_TOKEN:
+        results["telegram"] = send_telegram_alert.delay(message).get()
+    
+    if settings.DISCORD_WEBHOOK_URL:
+        embed = {
+            "title": f"🔴 PANIC DETECTED: ${symbol}",
+            "color": 0xff0000,  # Red
+            "fields": [
+                {"name": "Sentiment", "value": f"{sentiment:.1f}", "inline": True},
+                {"name": "Price Drop", "value": f"{price_drop:.1f}%", "inline": True},
+                {"name": "News Volume", "value": f"{news_count} articles", "inline": True},
+            ],
+            "footer": {"text": "MarketScanner • Watching for opportunity"},
+        }
+        results["discord"] = send_discord_alert.delay("", embed).get()
+    
+    return results

backend/app/workers/tasks/news_tasks.py

@@ -0,0 +1,108 @@
+"""
+News fetching tasks
+"""
+
+from datetime import datetime, timedelta
+import feedparser
+import structlog
+
+from app.workers.celery_app import celery_app
+from app.core.database import get_sync_db
+from app.core.config import settings
+
+logger = structlog.get_logger()
+
+# RSS Feeds to monitor
+NEWS_FEEDS = [
+    # General Financial News
+    {"name": "Yahoo Finance", "url": "https://finance.yahoo.com/news/rssindex"},
+    {"name": "Reuters Business", "url": "https://www.reutersagency.com/feed/?best-topics=business-finance&post_type=best"},
+    {"name": "CNBC", "url": "https://www.cnbc.com/id/100003114/device/rss/rss.html"},
+    {"name": "MarketWatch", "url": "https://feeds.marketwatch.com/marketwatch/topstories/"},
+    {"name": "Seeking Alpha", "url": "https://seekingalpha.com/market_currents.xml"},
+    {"name": "Bloomberg", "url": "https://www.bloomberg.com/feed/podcast/etf-report.xml"},
+    
+    # Tech
+    {"name": "TechCrunch", "url": "https://techcrunch.com/feed/"},
+    
+    # Crypto (because why not)
+    {"name": "CoinDesk", "url": "https://www.coindesk.com/arc/outboundfeeds/rss/"},
+]
+
+
+@celery_app.task(name="app.workers.tasks.news_tasks.fetch_all_news")
+def fetch_all_news():
+    """Fetch news from all configured sources."""
+    logger.info("Starting news fetch from all sources")
+    total_fetched = 0
+    
+    for feed_config in NEWS_FEEDS:
+        try:
+            count = fetch_from_feed(feed_config["name"], feed_config["url"])
+            total_fetched += count
+        except Exception as e:
+            logger.error(
+                "Failed to fetch from feed",
+                feed=feed_config["name"],
+                error=str(e)
+            )
+    
+    logger.info("News fetch complete", total_articles=total_fetched)
+    return {"fetched": total_fetched}
+
+
+@celery_app.task(name="app.workers.tasks.news_tasks.fetch_from_feed")
+def fetch_from_feed(source_name: str, feed_url: str) -> int:
+    """Fetch news from a single RSS feed."""
+    logger.info("Fetching from feed", source=source_name)
+    
+    try:
+        feed = feedparser.parse(feed_url)
+        articles_saved = 0
+        
+        for entry in feed.entries[:50]:  # Limit to 50 most recent
+            try:
+                # Extract data
+                title = entry.get("title", "")
+                url = entry.get("link", "")
+                summary = entry.get("summary", "")
+                author = entry.get("author", "")
+                
+                # Parse published date
+                published = entry.get("published_parsed") or entry.get("updated_parsed")
+                if published:
+                    published_at = datetime(*published[:6])
+                else:
+                    published_at = datetime.utcnow()
+                
+                # Save to database (skip if exists)
+                # This is a placeholder - actual implementation would use the db session
+                articles_saved += 1
+                
+            except Exception as e:
+                logger.warning(
+                    "Failed to process article",
+                    title=entry.get("title", "unknown"),
+                    error=str(e)
+                )
+        
+        logger.info("Feed processed", source=source_name, articles=articles_saved)
+        return articles_saved
+        
+    except Exception as e:
+        logger.error("Failed to parse feed", source=source_name, error=str(e))
+        return 0
+
+
+@celery_app.task(name="app.workers.tasks.news_tasks.cleanup_old_news")
+def cleanup_old_news(days: int = 90):
+    """Remove news articles older than specified days."""
+    logger.info("Starting news cleanup", days_to_keep=days)
+    
+    cutoff = datetime.utcnow() - timedelta(days=days)
+    
+    # Placeholder - actual implementation would delete from database
+    deleted_count = 0
+    
+    logger.info("News cleanup complete", deleted=deleted_count)
+    return {"deleted": deleted_count}

backend/app/workers/tasks/pattern_tasks.py

@@ -0,0 +1,142 @@
+"""
+Pattern detection and buy signal generation tasks
+"""
+
+from datetime import datetime, timedelta
+import structlog
+
+from app.workers.celery_app import celery_app
+from app.core.config import settings
+
+logger = structlog.get_logger()
+
+
+@celery_app.task(name="app.workers.tasks.pattern_tasks.detect_panic_events")
+def detect_panic_events():
+    """Detect new panic events based on sentiment and price drops."""
+    logger.info("Starting panic event detection")
+    
+    # Detection criteria:
+    # 1. Sentiment score drops below threshold
+    # 2. Price drops significantly (>5% in 24h)
+    # 3. News volume spikes
+    
+    # Placeholder - actual implementation would:
+    # - Query recent news sentiment by stock
+    # - Check price movements
+    # - Create panic_events records
+    
+    detected_count = 0
+    
+    logger.info("Panic detection complete", detected=detected_count)
+    return {"detected": detected_count}
+
+
+@celery_app.task(name="app.workers.tasks.pattern_tasks.generate_buy_signals")
+def generate_buy_signals():
+    """Generate buy signals based on historical patterns."""
+    logger.info("Starting buy signal generation")
+    
+    # Signal generation criteria:
+    # 1. Active panic event exists
+    # 2. Similar historical events had good recovery
+    # 3. Price is near or past typical bottom
+    # 4. Volume indicates capitulation
+    
+    # Placeholder - actual implementation would:
+    # - Find stocks with active panic events
+    # - Match against historical patterns
+    # - Calculate confidence scores
+    # - Create buy_signals records
+    
+    signals_count = 0
+    
+    logger.info("Signal generation complete", signals=signals_count)
+    return {"generated": signals_count}
+
+
+@celery_app.task(name="app.workers.tasks.pattern_tasks.analyze_historical_pattern")
+def analyze_historical_pattern(stock_id: str, event_type: str):
+    """Analyze historical patterns for a specific stock and event type."""
+    logger.info("Analyzing historical pattern", stock_id=stock_id, event_type=event_type)
+    
+    # Would query past panic events for this stock
+    # Calculate statistics:
+    # - Average/median drawdown
+    # - Average/median recovery time
+    # - Average/median recovery percentage
+    # - Success rate (how often did it recover)
+    
+    return {
+        "stock_id": stock_id,
+        "event_type": event_type,
+        "pattern": None,  # Would contain pattern data
+    }
+
+
+@celery_app.task(name="app.workers.tasks.pattern_tasks.calculate_confidence_score")
+def calculate_confidence_score(
+    stock_id: str,
+    current_drawdown: float,
+    current_sentiment: float,
+    historical_pattern: dict,
+) -> float:
+    """Calculate confidence score for a potential buy signal."""
+    
+    # Factors:
+    # 1. How close is current drawdown to historical average
+    # 2. How negative is sentiment (capitulation indicator)
+    # 3. Pattern reliability (sample size, consistency)
+    # 4. Market conditions (sector performance, overall market)
+    
+    score = 0.5  # Base score
+    
+    # Adjust based on drawdown match
+    if historical_pattern and historical_pattern.get("avg_drawdown"):
+        avg_drawdown = historical_pattern["avg_drawdown"]
+        drawdown_ratio = current_drawdown / avg_drawdown
+        if 0.8 <= drawdown_ratio <= 1.2:
+            score += 0.2  # Close to historical average
+    
+    # Adjust based on sentiment (more panic = higher score)
+    if current_sentiment < settings.PANIC_THRESHOLD:
+        panic_intensity = abs(current_sentiment - settings.PANIC_THRESHOLD) / 50
+        score += min(panic_intensity * 0.2, 0.2)
+    
+    # Adjust based on pattern reliability
+    if historical_pattern and historical_pattern.get("event_count", 0) >= 3:
+        score += 0.1  # Multiple historical examples
+    
+    return min(max(score, 0), 1)  # Clamp to 0-1
+
+
+@celery_app.task(name="app.workers.tasks.pattern_tasks.update_panic_event_status")
+def update_panic_event_status():
+    """Update panic events - check if they've ended/recovered."""
+    logger.info("Updating panic event statuses")
+    
+    # Check active (incomplete) panic events
+    # Mark as complete if:
+    # - Price has recovered to pre-panic levels
+    # - Sentiment has normalized
+    # - Enough time has passed
+    
+    updated_count = 0
+    
+    logger.info("Panic status update complete", updated=updated_count)
+    return {"updated": updated_count}
+
+
+@celery_app.task(name="app.workers.tasks.pattern_tasks.rebuild_patterns")
+def rebuild_patterns(stock_id: str = None):
+    """Rebuild historical patterns from panic events."""
+    logger.info("Rebuilding patterns", stock_id=stock_id or "all")
+    
+    # Aggregate all completed panic events
+    # Group by stock and event type
+    # Calculate pattern statistics
+    
+    patterns_count = 0
+    
+    logger.info("Pattern rebuild complete", patterns=patterns_count)
+    return {"rebuilt": patterns_count}

backend/app/workers/tasks/sentiment_tasks.py

@@ -0,0 +1,137 @@
+"""
+Sentiment analysis tasks
+"""
+
+from typing import Optional
+import structlog
+from openai import OpenAI
+
+from app.workers.celery_app import celery_app
+from app.core.config import settings
+
+logger = structlog.get_logger()
+
+
+def get_openai_client() -> Optional[OpenAI]:
+    """Get OpenAI client if configured."""
+    if settings.OPENAI_API_KEY:
+        return OpenAI(api_key=settings.OPENAI_API_KEY)
+    return None
+
+
+@celery_app.task(name="app.workers.tasks.sentiment_tasks.process_unanalyzed_news")
+def process_unanalyzed_news():
+    """Process all news articles that haven't been sentiment analyzed."""
+    logger.info("Starting sentiment analysis batch")
+    
+    # Placeholder - would query database for unprocessed articles
+    processed_count = 0
+    
+    logger.info("Sentiment analysis complete", processed=processed_count)
+    return {"processed": processed_count}
+
+
+@celery_app.task(name="app.workers.tasks.sentiment_tasks.analyze_sentiment")
+def analyze_sentiment(article_id: str, title: str, content: str):
+    """Analyze sentiment of a single article using OpenAI."""
+    logger.info("Analyzing sentiment", article_id=article_id)
+    
+    client = get_openai_client()
+    if not client:
+        logger.warning("OpenAI not configured, using fallback")
+        return fallback_sentiment_analysis(title, content)
+    
+    try:
+        # Prepare text (limit length)
+        text = f"Title: {title}\n\nContent: {content[:2000]}"
+        
+        response = client.chat.completions.create(
+            model=settings.OPENAI_MODEL,
+            messages=[
+                {
+                    "role": "system",
+                    "content": """You are a financial sentiment analyzer. Analyze the given news article and respond with a JSON object containing:
+                    - score: a number from -100 (extremely negative/panic) to +100 (extremely positive/euphoric)
+                    - label: one of "negative", "neutral", or "positive"
+                    - confidence: a number from 0 to 1 indicating confidence in the analysis
+                    - stocks: list of stock symbols mentioned (if any)
+                    - summary: one-sentence summary of the sentiment
+                    
+                    Focus on:
+                    - Financial impact
+                    - Market reaction implications
+                    - Panic/fear indicators
+                    - Earnings/guidance implications
+                    """
+                },
+                {
+                    "role": "user",
+                    "content": text
+                }
+            ],
+            response_format={"type": "json_object"},
+            temperature=0.3,
+        )
+        
+        result = response.choices[0].message.content
+        logger.info("Sentiment analyzed", article_id=article_id, result=result)
+        return result
+        
+    except Exception as e:
+        logger.error("Sentiment analysis failed", article_id=article_id, error=str(e))
+        return fallback_sentiment_analysis(title, content)
+
+
+def fallback_sentiment_analysis(title: str, content: str) -> dict:
+    """Simple keyword-based sentiment analysis as fallback."""
+    text = f"{title} {content}".lower()
+    
+    negative_words = [
+        "crash", "plunge", "collapse", "scandal", "fraud", "lawsuit",
+        "investigation", "bankruptcy", "layoffs", "miss", "decline",
+        "warning", "downgrade", "sell", "bear", "crisis", "fear",
+        "panic", "loss", "debt", "default", "recession"
+    ]
+    
+    positive_words = [
+        "surge", "rally", "growth", "profit", "beat", "upgrade",
+        "buy", "bull", "record", "breakout", "opportunity",
+        "dividend", "expansion", "innovation", "deal", "acquisition"
+    ]
+    
+    neg_count = sum(1 for word in negative_words if word in text)
+    pos_count = sum(1 for word in positive_words if word in text)
+    
+    total = neg_count + pos_count
+    if total == 0:
+        score = 0
+        label = "neutral"
+    else:
+        score = ((pos_count - neg_count) / total) * 100
+        if score < -20:
+            label = "negative"
+        elif score > 20:
+            label = "positive"
+        else:
+            label = "neutral"
+    
+    return {
+        "score": round(score, 2),
+        "label": label,
+        "confidence": min(0.3 + (total * 0.1), 0.7),  # Low confidence for fallback
+        "stocks": [],
+        "summary": "Analyzed using keyword matching (fallback)",
+    }
+
+
+@celery_app.task(name="app.workers.tasks.sentiment_tasks.batch_analyze")
+def batch_analyze(article_ids: list):
+    """Analyze multiple articles in batch."""
+    logger.info("Starting batch analysis", count=len(article_ids))
+    
+    results = []
+    for article_id in article_ids:
+        # Would fetch article from database and analyze
+        results.append({"article_id": article_id, "status": "pending"})
+    
+    return {"analyzed": len(results), "results": results}

backend/app/workers/tasks/stock_tasks.py

@@ -0,0 +1,102 @@
+"""
+Stock data fetching tasks
+"""
+
+import yfinance as yf
+import structlog
+
+from app.workers.celery_app import celery_app
+from app.core.config import settings
+
+logger = structlog.get_logger()
+
+
+@celery_app.task(name="app.workers.tasks.stock_tasks.update_stock_prices")
+def update_stock_prices():
+    """Update prices for all tracked stocks."""
+    logger.info("Starting stock price update")
+    
+    # Placeholder - would get active stocks from database
+    # For now, just demonstrate the concept
+    
+    updated_count = 0
+    
+    logger.info("Stock prices updated", count=updated_count)
+    return {"updated": updated_count}
+
+
+@celery_app.task(name="app.workers.tasks.stock_tasks.fetch_stock_price")
+def fetch_stock_price(symbol: str):
+    """Fetch current price for a single stock."""
+    logger.info("Fetching price", symbol=symbol)
+    
+    try:
+        ticker = yf.Ticker(symbol)
+        info = ticker.info
+        
+        return {
+            "symbol": symbol,
+            "price": info.get("currentPrice") or info.get("regularMarketPrice"),
+            "previous_close": info.get("previousClose"),
+            "volume": info.get("volume"),
+            "market_cap": info.get("marketCap"),
+        }
+    except Exception as e:
+        logger.error("Failed to fetch price", symbol=symbol, error=str(e))
+        return None
+
+
+@celery_app.task(name="app.workers.tasks.stock_tasks.fetch_historical_data")
+def fetch_historical_data(symbol: str, period: str = "10y"):
+    """Fetch historical price data for a stock."""
+    logger.info("Fetching historical data", symbol=symbol, period=period)
+    
+    try:
+        ticker = yf.Ticker(symbol)
+        hist = ticker.history(period=period)
+        
+        # Convert to list of dicts for storage
+        records = []
+        for idx, row in hist.iterrows():
+            records.append({
+                "time": idx.isoformat(),
+                "open": row["Open"],
+                "high": row["High"],
+                "low": row["Low"],
+                "close": row["Close"],
+                "volume": row["Volume"],
+            })
+        
+        logger.info(
+            "Historical data fetched",
+            symbol=symbol,
+            records=len(records)
+        )
+        return {"symbol": symbol, "records": len(records)}
+        
+    except Exception as e:
+        logger.error("Failed to fetch historical data", symbol=symbol, error=str(e))
+        return None
+
+
+@celery_app.task(name="app.workers.tasks.stock_tasks.update_stock_info")
+def update_stock_info(symbol: str):
+    """Update stock metadata (sector, industry, market cap, etc.)."""
+    logger.info("Updating stock info", symbol=symbol)
+    
+    try:
+        ticker = yf.Ticker(symbol)
+        info = ticker.info
+        
+        return {
+            "symbol": symbol,
+            "name": info.get("longName") or info.get("shortName"),
+            "sector": info.get("sector"),
+            "industry": info.get("industry"),
+            "market_cap": info.get("marketCap"),
+            "exchange": info.get("exchange"),
+            "country": info.get("country"),
+        }
+    except Exception as e:
+        logger.error("Failed to update stock info", symbol=symbol, error=str(e))
+        return None

backend/requirements.txt

@@ -0,0 +1,70 @@
+# FastAPI & Web
+fastapi==0.109.2
+uvicorn[standard]==0.27.1
+python-multipart==0.0.9
+python-jose[cryptography]==3.3.0
+passlib[bcrypt]==1.7.4
+httpx==0.26.0
+aiohttp==3.9.3
+websockets==12.0
+
+# Database
+sqlalchemy==2.0.25
+asyncpg==0.29.0
+psycopg2-binary==2.9.9
+alembic==1.13.1
+
+# Redis & Caching
+redis==5.0.1
+aioredis==2.0.1
+
+# Celery & Task Queue
+celery==5.3.6
+flower==2.0.1
+
+# Data Processing
+pandas==2.2.0
+numpy==1.26.3
+scipy==1.12.0
+
+# Stock Data
+yfinance==0.2.36
+alpha-vantage==2.3.1
+finnhub-python==2.4.19
+
+# News & Web Scraping
+feedparser==6.0.11
+newspaper3k==0.2.8
+beautifulsoup4==4.12.3
+lxml==5.1.0
+
+# NLP & Sentiment
+openai==1.12.0
+tiktoken==0.5.2
+nltk==3.8.1
+textblob==0.17.1
+transformers==4.37.2
+torch==2.2.0
+
+# Validation & Serialization
+pydantic==2.6.1
+pydantic-settings==2.1.0
+
+# Utilities
+python-dotenv==1.0.1
+structlog==24.1.0
+tenacity==8.2.3
+python-dateutil==2.8.2
+pytz==2024.1
+
+# Testing
+pytest==8.0.0
+pytest-asyncio==0.23.4
+pytest-cov==4.1.0
+httpx==0.26.0
+
+# Development
+black==24.1.1
+isort==5.13.2
+flake8==7.0.0
+mypy==1.8.0