Spaces:
Sleeping
Sleeping
File size: 4,940 Bytes
e6021a3 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 | """
Pattern Intelligence Router.
Endpoints for candlestick pattern detection, ML-based prediction,
pattern catalog, and historical accuracy analysis.
"""
from __future__ import annotations
import logging
from typing import List, Optional
from fastapi import APIRouter, Depends, HTTPException
from pydantic import BaseModel, Field
from app.dependencies import get_current_user
from app.models.user import User
logger = logging.getLogger(__name__)
router = APIRouter(prefix="/patterns", tags=["Pattern Intelligence"])
# ── Schemas ──────────────────────────────────────────────────────────────
class PatternAnalyzeRequest(BaseModel):
ticker: str = Field(..., min_length=1, max_length=20)
period: str = Field("2y", description="Historical data period")
horizon: int = Field(5, ge=1, le=30, description="Prediction horizon in days")
class MultiAnalyzeRequest(BaseModel):
tickers: List[str] = Field(..., min_items=1, max_items=10)
period: str = Field("2y")
horizon: int = Field(5, ge=1, le=30)
class BacktestAccuracyRequest(BaseModel):
ticker: str = Field(..., min_length=1, max_length=20)
period: str = Field("5y")
horizon: int = Field(5, ge=1, le=30)
# ── Endpoints ────────────────────────────────────────────────────────────
@router.post("/analyze")
async def analyze_patterns(
data: PatternAnalyzeRequest,
user: User = Depends(get_current_user),
):
"""
Detect candlestick patterns and predict price direction using
pattern-aware LightGBM model with advanced mathematical features.
Returns:
- Predicted direction (strong_up / neutral / strong_down)
- Confidence and probability distribution
- Detected patterns with reliability scores
- Top feature importances
- Advanced feature values (Hurst, Fractal, Entropy, etc.)
"""
from app.services.ml.pattern_recognition.predictor import predict_with_patterns
try:
result = await predict_with_patterns(
ticker=data.ticker,
period=data.period,
horizon=data.horizon,
)
return result
except ValueError as e:
raise HTTPException(status_code=400, detail=str(e))
except Exception as e:
logger.error("Pattern analysis failed for %s: %s", data.ticker, e, exc_info=True)
raise HTTPException(status_code=500, detail="Pattern analysis failed")
@router.post("/multi-analyze")
async def multi_analyze(
data: MultiAnalyzeRequest,
user: User = Depends(get_current_user),
):
"""Analyze multiple tickers and return comparative results."""
from app.services.ml.pattern_recognition.predictor import analyze_multiple
try:
results = await analyze_multiple(
tickers=data.tickers,
period=data.period,
horizon=data.horizon,
)
return results
except Exception as e:
logger.error("Multi-analysis failed: %s", e, exc_info=True)
raise HTTPException(status_code=500, detail="Multi-analysis failed")
@router.get("/catalog")
async def pattern_catalog(
user: User = Depends(get_current_user),
):
"""
Get the full catalog of all 35+ supported candlestick
and chart patterns with descriptions and reliability ratings.
"""
from app.services.ml.pattern_recognition.pattern_detector import pattern_detector
return {
"total_patterns": len(pattern_detector.get_pattern_catalog()),
"patterns": pattern_detector.get_pattern_catalog(),
}
@router.post("/backtest-accuracy")
async def backtest_accuracy(
data: BacktestAccuracyRequest,
user: User = Depends(get_current_user),
):
"""
Backtest pattern detection accuracy on historical data.
For each pattern type, returns occurrences, win rate,
average return, and actual vs theoretical reliability.
"""
from app.services.ml.pattern_recognition.predictor import backtest_pattern_accuracy
try:
result = await backtest_pattern_accuracy(
ticker=data.ticker,
period=data.period,
horizon=data.horizon,
)
return result
except ValueError as e:
raise HTTPException(status_code=400, detail=str(e))
except Exception as e:
logger.error("Backtest accuracy failed for %s: %s", data.ticker, e, exc_info=True)
raise HTTPException(status_code=500, detail="Backtest accuracy failed")
@router.post("/clear-cache")
async def clear_pattern_cache(
user: User = Depends(get_current_user),
):
"""Clear the pattern predictor model cache."""
from app.services.ml.pattern_recognition.predictor import clear_cache
count = clear_cache()
return {"cleared": count}
|