456 lines
14 KiB
Python
456 lines
14 KiB
Python
"""
|
|
Phase 3: Advanced Analytics API Endpoints
|
|
Performance tracking, pattern analysis, and reporting
|
|
"""
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException, Query
|
|
from sqlalchemy.orm import Session
|
|
from sqlalchemy import func
|
|
from datetime import datetime, date, timedelta
|
|
from typing import List, Optional
|
|
from app.db.database import get_db
|
|
from app.models.models import (
|
|
PerformanceSnapshot, TradePattern, LessonLearned, MonthlyReview, Trade
|
|
)
|
|
from app.schemas.schemas import (
|
|
PerformanceSnapshotCreate, PerformanceSnapshotResponse,
|
|
TradePatternCreate, TradePatternResponse,
|
|
LessonLearnedCreate, LessonLearnedResponse,
|
|
MonthlyReviewCreate, MonthlyReviewResponse
|
|
)
|
|
|
|
router = APIRouter(prefix="/api/analytics", tags=["Advanced Analytics"])
|
|
|
|
|
|
# ============================================================================
|
|
# PERFORMANCE SNAPSHOTS
|
|
# ============================================================================
|
|
|
|
@router.post("/snapshots", response_model=PerformanceSnapshotResponse, status_code=201)
|
|
async def create_performance_snapshot(
|
|
snapshot: PerformanceSnapshotCreate,
|
|
db: Session = Depends(get_db)
|
|
):
|
|
"""Create a performance snapshot"""
|
|
db_snapshot = PerformanceSnapshot(**snapshot.dict())
|
|
db.add(db_snapshot)
|
|
db.commit()
|
|
db.refresh(db_snapshot)
|
|
return db_snapshot
|
|
|
|
|
|
@router.get("/snapshots", response_model=List[PerformanceSnapshotResponse])
|
|
async def list_performance_snapshots(
|
|
start_date: Optional[str] = Query(None),
|
|
end_date: Optional[str] = Query(None),
|
|
limit: int = Query(30, ge=1, le=365),
|
|
db: Session = Depends(get_db)
|
|
):
|
|
"""List performance snapshots with optional date range"""
|
|
query = db.query(PerformanceSnapshot)
|
|
|
|
if start_date:
|
|
start = datetime.fromisoformat(start_date).date()
|
|
query = query.filter(PerformanceSnapshot.snapshot_date >= start)
|
|
|
|
if end_date:
|
|
end = datetime.fromisoformat(end_date).date()
|
|
query = query.filter(PerformanceSnapshot.snapshot_date <= end)
|
|
|
|
return query.order_by(PerformanceSnapshot.snapshot_date.desc()).limit(limit).all()
|
|
|
|
|
|
@router.get("/snapshots/stats/monthly")
|
|
async def get_monthly_stats(
|
|
year: int = Query(...),
|
|
month: int = Query(..., ge=1, le=12),
|
|
db: Session = Depends(get_db)
|
|
):
|
|
"""Get monthly aggregate statistics"""
|
|
snapshots = db.query(PerformanceSnapshot).filter(
|
|
func.extract('year', PerformanceSnapshot.snapshot_date) == year,
|
|
func.extract('month', PerformanceSnapshot.snapshot_date) == month
|
|
).all()
|
|
|
|
if not snapshots:
|
|
return {
|
|
"year": year,
|
|
"month": month,
|
|
"trading_days": 0,
|
|
"total_pnl": 0.0,
|
|
"avg_daily_pnl": 0.0,
|
|
"best_day_pnl": 0.0,
|
|
"worst_day_pnl": 0.0,
|
|
"win_rate": 0.0,
|
|
"total_trades": 0
|
|
}
|
|
|
|
total_pnl = sum(s.daily_pnl for s in snapshots)
|
|
total_trades = sum(s.total_trades for s in snapshots)
|
|
winning_days = sum(1 for s in snapshots if s.daily_pnl > 0)
|
|
trading_days = len(snapshots)
|
|
|
|
return {
|
|
"year": year,
|
|
"month": month,
|
|
"trading_days": trading_days,
|
|
"total_pnl": total_pnl,
|
|
"avg_daily_pnl": total_pnl / trading_days if trading_days > 0 else 0,
|
|
"best_day_pnl": max((s.daily_pnl for s in snapshots), default=0),
|
|
"worst_day_pnl": min((s.daily_pnl for s in snapshots), default=0),
|
|
"win_rate": (winning_days / trading_days * 100) if trading_days > 0 else 0,
|
|
"total_trades": total_trades,
|
|
"winning_days": winning_days,
|
|
"losing_days": trading_days - winning_days
|
|
}
|
|
|
|
|
|
@router.get("/snapshots/stats/yearly")
|
|
async def get_yearly_stats(
|
|
year: int = Query(...),
|
|
db: Session = Depends(get_db)
|
|
):
|
|
"""Get yearly aggregate statistics"""
|
|
snapshots = db.query(PerformanceSnapshot).filter(
|
|
func.extract('year', PerformanceSnapshot.snapshot_date) == year
|
|
).all()
|
|
|
|
if not snapshots:
|
|
return {"year": year, "message": "No data for this year"}
|
|
|
|
total_pnl = sum(s.daily_pnl for s in snapshots)
|
|
total_trades = sum(s.total_trades for s in snapshots)
|
|
winning_days = sum(1 for s in snapshots if s.daily_pnl > 0)
|
|
trading_days = len(snapshots)
|
|
|
|
return {
|
|
"year": year,
|
|
"trading_days": trading_days,
|
|
"total_pnl": total_pnl,
|
|
"avg_daily_pnl": total_pnl / trading_days if trading_days > 0 else 0,
|
|
"best_day": max((s.daily_pnl for s in snapshots), default=0),
|
|
"worst_day": min((s.daily_pnl for s in snapshots), default=0),
|
|
"win_rate": (winning_days / trading_days * 100) if trading_days > 0 else 0,
|
|
"total_trades": total_trades,
|
|
"best_month": None, # Can be calculated from monthly stats
|
|
"worst_month": None
|
|
}
|
|
|
|
|
|
# ============================================================================
|
|
# TRADE PATTERNS
|
|
# ============================================================================
|
|
|
|
@router.post("/patterns", response_model=TradePatternResponse, status_code=201)
|
|
async def create_pattern(
|
|
pattern: TradePatternCreate,
|
|
db: Session = Depends(get_db)
|
|
):
|
|
"""Identify and create a new trade pattern"""
|
|
db_pattern = TradePattern(**pattern.dict())
|
|
db.add(db_pattern)
|
|
db.commit()
|
|
db.refresh(db_pattern)
|
|
return db_pattern
|
|
|
|
|
|
@router.get("/patterns", response_model=List[TradePatternResponse])
|
|
async def list_patterns(
|
|
min_confidence: float = Query(0, ge=0, le=100),
|
|
min_sample_count: int = Query(3, ge=1),
|
|
db: Session = Depends(get_db)
|
|
):
|
|
"""List identified trade patterns"""
|
|
patterns = db.query(TradePattern).filter(
|
|
TradePattern.confidence_score >= min_confidence,
|
|
TradePattern.sample_count >= min_sample_count
|
|
).order_by(TradePattern.confidence_score.desc()).all()
|
|
|
|
return patterns
|
|
|
|
|
|
@router.get("/patterns/{pattern_id}", response_model=TradePatternResponse)
|
|
async def get_pattern(
|
|
pattern_id: int,
|
|
db: Session = Depends(get_db)
|
|
):
|
|
"""Get specific pattern details"""
|
|
pattern = db.query(TradePattern).filter(TradePattern.id == pattern_id).first()
|
|
if not pattern:
|
|
raise HTTPException(status_code=404, detail="Pattern not found")
|
|
return pattern
|
|
|
|
|
|
@router.get("/patterns/stats/best")
|
|
async def get_best_patterns(
|
|
limit: int = Query(5, ge=1, le=20),
|
|
db: Session = Depends(get_db)
|
|
):
|
|
"""Get your top performing patterns"""
|
|
patterns = db.query(TradePattern).order_by(
|
|
TradePattern.confidence_score.desc()
|
|
).limit(limit).all()
|
|
|
|
return [
|
|
{
|
|
"pattern": p.pattern_name,
|
|
"win_rate": p.win_rate,
|
|
"confidence": p.confidence_score,
|
|
"sample_size": p.sample_count,
|
|
"total_profit": p.total_profit,
|
|
"best_timeframe": p.best_timeframe,
|
|
"best_time": p.best_time_of_day
|
|
}
|
|
for p in patterns
|
|
]
|
|
|
|
|
|
# ============================================================================
|
|
# LESSONS LEARNED
|
|
# ============================================================================
|
|
|
|
@router.post("/lessons", response_model=LessonLearnedResponse, status_code=201)
|
|
async def create_lesson(
|
|
lesson: LessonLearnedCreate,
|
|
db: Session = Depends(get_db)
|
|
):
|
|
"""Log a lesson learned"""
|
|
db_lesson = LessonLearned(**lesson.dict())
|
|
db.add(db_lesson)
|
|
db.commit()
|
|
db.refresh(db_lesson)
|
|
return db_lesson
|
|
|
|
|
|
@router.get("/lessons", response_model=List[LessonLearnedResponse])
|
|
async def list_lessons(
|
|
category: Optional[str] = Query(None),
|
|
importance: Optional[str] = Query(None),
|
|
tag: Optional[str] = Query(None),
|
|
limit: int = Query(20, ge=1, le=100),
|
|
db: Session = Depends(get_db)
|
|
):
|
|
"""List lessons learned with optional filters"""
|
|
query = db.query(LessonLearned).filter(LessonLearned.status == "active")
|
|
|
|
if category:
|
|
query = query.filter(LessonLearned.category == category)
|
|
if importance:
|
|
query = query.filter(LessonLearned.importance == importance)
|
|
|
|
lessons = query.order_by(LessonLearned.date_learned.desc()).limit(limit).all()
|
|
|
|
# Filter by tag if specified
|
|
if tag:
|
|
lessons = [l for l in lessons if tag in l.tags]
|
|
|
|
return lessons
|
|
|
|
|
|
@router.get("/lessons/categories")
|
|
async def get_lesson_categories(db: Session = Depends(get_db)):
|
|
"""Get available lesson categories"""
|
|
categories = db.query(LessonLearned.category).distinct().all()
|
|
return {
|
|
"categories": [c[0] for c in categories if c[0]],
|
|
"available": ["entry", "exit", "risk", "psychology", "market"]
|
|
}
|
|
|
|
|
|
@router.get("/lessons/recurring-mistakes")
|
|
async def get_recurring_mistakes(
|
|
limit: int = Query(10, ge=1, le=20),
|
|
db: Session = Depends(get_db)
|
|
):
|
|
"""Identify recurring mistakes from lessons"""
|
|
negative_lessons = db.query(LessonLearned).filter(
|
|
LessonLearned.impact == "negative"
|
|
).order_by(LessonLearned.date_learned.desc()).all()
|
|
|
|
# Count tag occurrences
|
|
tag_counts = {}
|
|
for lesson in negative_lessons:
|
|
for tag in lesson.tags:
|
|
tag_counts[tag] = tag_counts.get(tag, 0) + 1
|
|
|
|
# Sort by frequency
|
|
recurring = sorted(tag_counts.items(), key=lambda x: x[1], reverse=True)
|
|
|
|
return {
|
|
"recurring_mistakes": recurring[:limit],
|
|
"total_negative_lessons": len(negative_lessons),
|
|
"recommendation": "Focus on preventing these recurring mistakes"
|
|
}
|
|
|
|
|
|
# ============================================================================
|
|
# MONTHLY REVIEWS
|
|
# ============================================================================
|
|
|
|
@router.post("/reviews/monthly", response_model=MonthlyReviewResponse, status_code=201)
|
|
async def create_monthly_review(
|
|
review: MonthlyReviewCreate,
|
|
db: Session = Depends(get_db)
|
|
):
|
|
"""Create a monthly performance review"""
|
|
# Check if review already exists
|
|
existing = db.query(MonthlyReview).filter(
|
|
MonthlyReview.year == review.year,
|
|
MonthlyReview.month == review.month
|
|
).first()
|
|
|
|
if existing:
|
|
raise HTTPException(
|
|
status_code=400,
|
|
detail=f"Monthly review for {review.year}-{review.month} already exists"
|
|
)
|
|
|
|
db_review = MonthlyReview(**review.dict())
|
|
db.add(db_review)
|
|
db.commit()
|
|
db.refresh(db_review)
|
|
return db_review
|
|
|
|
|
|
@router.get("/reviews/monthly", response_model=List[MonthlyReviewResponse])
|
|
async def list_monthly_reviews(
|
|
year: Optional[int] = Query(None),
|
|
limit: int = Query(12, ge=1, le=60),
|
|
db: Session = Depends(get_db)
|
|
):
|
|
"""List monthly reviews"""
|
|
query = db.query(MonthlyReview)
|
|
|
|
if year:
|
|
query = query.filter(MonthlyReview.year == year)
|
|
|
|
return query.order_by(
|
|
MonthlyReview.year.desc(),
|
|
MonthlyReview.month.desc()
|
|
).limit(limit).all()
|
|
|
|
|
|
@router.get("/reviews/quarterly")
|
|
async def get_quarterly_review(
|
|
year: int = Query(...),
|
|
quarter: int = Query(..., ge=1, le=4),
|
|
db: Session = Depends(get_db)
|
|
):
|
|
"""Get quarterly performance review"""
|
|
months = {
|
|
1: [1, 2, 3],
|
|
2: [4, 5, 6],
|
|
3: [7, 8, 9],
|
|
4: [10, 11, 12]
|
|
}
|
|
|
|
month_list = months[quarter]
|
|
reviews = db.query(MonthlyReview).filter(
|
|
MonthlyReview.year == year,
|
|
MonthlyReview.month.in_(month_list)
|
|
).all()
|
|
|
|
if not reviews:
|
|
return {"quarter": quarter, "year": year, "message": "No data"}
|
|
|
|
total_pnl = sum(r.total_pnl for r in reviews)
|
|
total_trades = sum(r.total_trades for r in reviews)
|
|
avg_win_rate = sum(r.win_rate for r in reviews) / len(reviews) if reviews else 0
|
|
|
|
return {
|
|
"quarter": quarter,
|
|
"year": year,
|
|
"months_covered": month_list,
|
|
"total_pnl": total_pnl,
|
|
"total_trades": total_trades,
|
|
"avg_win_rate": avg_win_rate,
|
|
"best_month": max((r.total_pnl for r in reviews), default=0),
|
|
"worst_month": min((r.total_pnl for r in reviews), default=0),
|
|
"monthly_reviews": [
|
|
{
|
|
"month": r.month,
|
|
"pnl": r.total_pnl,
|
|
"win_rate": r.win_rate,
|
|
"trades": r.total_trades
|
|
}
|
|
for r in reviews
|
|
]
|
|
}
|
|
|
|
|
|
# ============================================================================
|
|
# COMPREHENSIVE ANALYTICS DASHBOARD
|
|
# ============================================================================
|
|
|
|
@router.get("/dashboard")
|
|
async def get_analytics_dashboard(
|
|
period: str = Query("month", regex="^(week|month|quarter|year)$"),
|
|
db: Session = Depends(get_db)
|
|
):
|
|
"""Get comprehensive analytics dashboard"""
|
|
today = date.today()
|
|
|
|
# Determine date range
|
|
if period == "week":
|
|
start_date = today - timedelta(days=7)
|
|
elif period == "month":
|
|
start_date = today - timedelta(days=30)
|
|
elif period == "quarter":
|
|
start_date = today - timedelta(days=90)
|
|
else: # year
|
|
start_date = today - timedelta(days=365)
|
|
|
|
# Get snapshots for period
|
|
snapshots = db.query(PerformanceSnapshot).filter(
|
|
PerformanceSnapshot.snapshot_date >= start_date
|
|
).all()
|
|
|
|
# Get patterns
|
|
patterns = db.query(TradePattern).order_by(
|
|
TradePattern.confidence_score.desc()
|
|
).limit(5).all()
|
|
|
|
# Get recent lessons
|
|
lessons = db.query(LessonLearned).filter(
|
|
LessonLearned.status == "active"
|
|
).order_by(LessonLearned.date_learned.desc()).limit(5).all()
|
|
|
|
# Calculate metrics
|
|
total_pnl = sum(s.daily_pnl for s in snapshots)
|
|
total_trades = sum(s.total_trades for s in snapshots)
|
|
winning_days = sum(1 for s in snapshots if s.daily_pnl > 0)
|
|
avg_win_rate = sum(s.win_rate for s in snapshots) / len(snapshots) if snapshots else 0
|
|
|
|
return {
|
|
"period": period,
|
|
"snapshot_count": len(snapshots),
|
|
"performance": {
|
|
"total_pnl": total_pnl,
|
|
"avg_daily_pnl": total_pnl / len(snapshots) if snapshots else 0,
|
|
"total_trades": total_trades,
|
|
"winning_days": winning_days,
|
|
"losing_days": len(snapshots) - winning_days,
|
|
"avg_win_rate": avg_win_rate,
|
|
"best_day": max((s.daily_pnl for s in snapshots), default=0),
|
|
"worst_day": min((s.daily_pnl for s in snapshots), default=0)
|
|
},
|
|
"top_patterns": [
|
|
{
|
|
"name": p.pattern_name,
|
|
"confidence": p.confidence_score,
|
|
"win_rate": p.win_rate,
|
|
"samples": p.sample_count
|
|
}
|
|
for p in patterns
|
|
],
|
|
"recent_lessons": [
|
|
{
|
|
"category": l.category,
|
|
"lesson": l.lesson_text[:100],
|
|
"importance": l.importance,
|
|
"date": l.date_learned.isoformat()
|
|
}
|
|
for l in lessons
|
|
]
|
|
}
|