""" ========================================= CV6 AI Learning Service Calculates performance analytics from trade history. Generates AI suggestions. ========================================= """ import time import uuid from collections import defaultdict from datetime import datetime, timedelta from typing import Any, Dict, List, Optional, Tuple from sqlalchemy.orm import Session from loguru import logger from app.database.session import SessionLocal from app.learning.learning_models import TradeRecord class LearningService: """ Reads all TradeRecord entries and derives: - Win rate overall + per strategy/style/AI - Best/worst strategy by P&L and win rate - AI accuracy per model - Best trading times - Best sectors - Monthly and yearly reports - Auto-improvement suggestions """ # ── Trade Recording ─────────────────────────────────────────────────────── def record_trade(self, trade_data: dict) -> str: """Save a completed trade to the learning DB. Returns trade ID.""" db: Session = SessionLocal() try: trade_id = str(uuid.uuid4())[:16] rec = TradeRecord( id = trade_id, symbol = trade_data.get("symbol", ""), sector = trade_data.get("sector", "Other"), style = trade_data.get("style", "INTRADAY"), broker = trade_data.get("broker", ""), side = trade_data.get("side", "BUY"), qty = trade_data.get("qty", 0), entry_price = trade_data.get("entry_price", 0.0), exit_price = trade_data.get("exit_price", 0.0), stop_loss = trade_data.get("stop_loss", 0.0), target_price = trade_data.get("target_price", 0.0), pnl = trade_data.get("pnl", 0.0), pnl_pct = trade_data.get("pnl_pct", 0.0), entry_time = trade_data.get("entry_time", ""), exit_time = trade_data.get("exit_time", datetime.now().strftime("%Y-%m-%d %H:%M:%S")), holding_minutes = trade_data.get("holding_minutes", 0.0), exit_reason = trade_data.get("exit_reason", "MANUAL"), strategy = trade_data.get("strategy", ""), ai_models_used = trade_data.get("ai_models_used", ""), ai_signal = trade_data.get("ai_signal", ""), ai_confidence = trade_data.get("ai_confidence", 0.0), ai_correct = bool(trade_data.get("ai_correct", False)), ai_reason = trade_data.get("ai_reason", ""), from_cache = bool(trade_data.get("from_cache", False)), market_regime = trade_data.get("market_regime", "UNKNOWN"), nifty_pct = trade_data.get("nifty_pct", 0.0), vix = trade_data.get("vix", 0.0), news_sentiment = trade_data.get("news_sentiment", "NEUTRAL"), slippage_pct = trade_data.get("slippage_pct", 0.0), mode = trade_data.get("mode", "PAPER"), created_at = datetime.now().strftime("%Y-%m-%d %H:%M:%S"), ) indicators = trade_data.get("indicators", {}) if indicators: rec.set_indicators(indicators) db.add(rec) db.commit() return trade_id except Exception as e: db.rollback() logger.error(f"[Learning] Failed to record trade: {e}") return "" finally: db.close() # ── Analytics ───────────────────────────────────────────────────────────── def get_stats(self) -> dict: """Overall performance statistics.""" db = SessionLocal() try: records = db.query(TradeRecord).all() if not records: return self._empty_stats() return self._calc_stats(records) finally: db.close() def get_strategy_stats(self) -> List[dict]: """Performance breakdown per strategy.""" db = SessionLocal() try: records = db.query(TradeRecord).all() return self._group_stats(records, key="strategy") finally: db.close() def get_style_stats(self) -> List[dict]: """Performance breakdown per trading style.""" db = SessionLocal() try: records = db.query(TradeRecord).all() return self._group_stats(records, key="style") finally: db.close() def get_ai_accuracy(self) -> List[dict]: """AI model accuracy analysis.""" db = SessionLocal() try: records = db.query(TradeRecord).filter(TradeRecord.ai_signal != "").all() total = len(records) correct = sum(1 for r in records if r.ai_correct) accuracy = round(correct / total * 100, 1) if total > 0 else 0 avg_conf = round(sum(r.ai_confidence or 0 for r in records) / max(total, 1), 1) cache_hits = sum(1 for r in records if r.from_cache) return [{ "metric": "Overall AI Accuracy", "total_signals": total, "correct": correct, "accuracy_pct": accuracy, "avg_confidence": avg_conf, "cache_hits": cache_hits, "cache_hit_rate": round(cache_hits / max(total, 1) * 100, 1), }] finally: db.close() def get_time_analysis(self) -> List[dict]: """Best/worst trading hours.""" db = SessionLocal() try: records = db.query(TradeRecord).filter(TradeRecord.entry_time != None).all() hourly: Dict[int, List[float]] = defaultdict(list) for r in records: try: hour = int(r.entry_time.split(" ")[1].split(":")[0]) hourly[hour].append(r.pnl or 0) except Exception: pass result = [] for hour in sorted(hourly.keys()): pnls = hourly[hour] wins = sum(1 for p in pnls if p > 0) result.append({ "hour": hour, "label": f"{hour:02d}:00", "trades": len(pnls), "win_rate": round(wins / len(pnls) * 100, 1), "avg_pnl": round(sum(pnls) / len(pnls), 2), "total_pnl": round(sum(pnls), 2), }) return result finally: db.close() def get_sector_stats(self) -> List[dict]: """Best/worst sectors.""" db = SessionLocal() try: records = db.query(TradeRecord).filter(TradeRecord.sector != None).all() return self._group_stats(records, key="sector") finally: db.close() def get_monthly_report(self, year: int, month: int) -> dict: """P&L, win rate, drawdown for a specific month.""" db = SessionLocal() try: prefix = f"{year}-{month:02d}" records = db.query(TradeRecord).filter( TradeRecord.exit_time.like(f"{prefix}%") ).all() stats = self._calc_stats(records) stats["year"] = year stats["month"] = month stats["label"] = f"{year}-{month:02d}" return stats finally: db.close() def get_monthly_series(self) -> List[dict]: """Monthly P&L series for all recorded months.""" db = SessionLocal() try: records = db.query(TradeRecord).filter( TradeRecord.exit_time != None ).all() monthly: Dict[str, List[TradeRecord]] = defaultdict(list) for r in records: try: key = r.exit_time[:7] # "YYYY-MM" monthly[key].append(r) except Exception: pass result = [] for key in sorted(monthly.keys()): stats = self._calc_stats(monthly[key]) stats["month"] = key result.append(stats) return result finally: db.close() def get_suggestions(self) -> List[dict]: """ Auto-generate improvement suggestions based on trade data. """ db = SessionLocal() try: records = db.query(TradeRecord).all() return self._generate_suggestions(records) finally: db.close() def get_recent_trades(self, limit: int = 50) -> List[dict]: db = SessionLocal() try: records = ( db.query(TradeRecord) .order_by(TradeRecord.created_at.desc()) .limit(limit) .all() ) return [r.to_dict() for r in records] finally: db.close() # ── Internal helpers ────────────────────────────────────────────────────── def _calc_stats(self, records: list) -> dict: if not records: return self._empty_stats() total = len(records) wins = [r for r in records if (r.pnl or 0) > 0] losses = [r for r in records if (r.pnl or 0) <= 0] win_rate = round(len(wins) / total * 100, 1) pnls = [r.pnl or 0 for r in records] total_pnl = round(sum(pnls), 2) avg_pnl = round(sum(pnls) / total, 2) avg_win = round(sum(r.pnl or 0 for r in wins) / max(len(wins), 1), 2) avg_loss = round(sum(r.pnl or 0 for r in losses) / max(len(losses), 1), 2) rr_ratio = round(abs(avg_win / avg_loss), 2) if avg_loss != 0 else 0 # Max drawdown (running peak) running_pnl = 0.0 peak = 0.0 max_dd = 0.0 for p in pnls: running_pnl += p if running_pnl > peak: peak = running_pnl dd = peak - running_pnl if dd > max_dd: max_dd = dd return { "total_trades": total, "wins": len(wins), "losses": len(losses), "win_rate": win_rate, "total_pnl": total_pnl, "avg_pnl": avg_pnl, "avg_win": avg_win, "avg_loss": avg_loss, "risk_reward": rr_ratio, "max_drawdown": round(max_dd, 2), "profit_factor": round(sum(r.pnl or 0 for r in wins) / max(abs(sum(r.pnl or 0 for r in losses)), 1), 2), } def _group_stats(self, records: list, key: str) -> List[dict]: grouped: Dict[str, list] = defaultdict(list) for r in records: k = getattr(r, key, "Unknown") or "Unknown" grouped[k].append(r) result = [] for k, recs in grouped.items(): stats = self._calc_stats(recs) stats[key] = k result.append(stats) result.sort(key=lambda x: x["total_pnl"], reverse=True) return result def _generate_suggestions(self, records: list) -> List[dict]: suggestions = [] if not records: suggestions.append({ "type": "INFO", "priority": "LOW", "title": "No trade data yet", "message": "Start the autonomous engine in PAPER mode to build your learning dataset.", }) return suggestions stats = self._calc_stats(records) win_rate = stats["win_rate"] rr = stats["risk_reward"] if win_rate < 45: suggestions.append({ "type": "WARNING", "priority": "HIGH", "title": "Low Win Rate", "message": f"Win rate is {win_rate}%. Consider raising AI confidence threshold to 75%+ to filter weak signals.", "action": "Increase consensus_threshold", }) if rr < 1.0: suggestions.append({ "type": "WARNING", "priority": "HIGH", "title": "Unfavorable Risk/Reward", "message": f"R:R is {rr}. Tighten stop losses or widen targets.", "action": "Adjust SL/TP ratios per style", }) # Best strategy strat_stats = self._group_stats(records, "strategy") if strat_stats: best = max(strat_stats, key=lambda x: x["win_rate"]) worst = min(strat_stats, key=lambda x: x["win_rate"]) if best["win_rate"] > 55: suggestions.append({ "type": "OPPORTUNITY", "priority": "MEDIUM", "title": f"Top Strategy: {best['strategy']}", "message": f"{best['strategy']} has {best['win_rate']}% win rate. Consider allocating more capital to it.", "action": f"Increase weight for {best['strategy']}", }) if worst["win_rate"] < 40 and worst["total_trades"] >= 5: suggestions.append({ "type": "WARNING", "priority": "MEDIUM", "title": f"Weak Strategy: {worst['strategy']}", "message": f"{worst['strategy']} has only {worst['win_rate']}% win rate. Consider disabling it.", "action": f"Disable {worst['strategy']} in strategies_enabled", }) # AI accuracy ai_records = [r for r in records if r.ai_signal] if ai_records: ai_correct = sum(1 for r in ai_records if r.ai_correct) ai_acc = round(ai_correct / len(ai_records) * 100, 1) if ai_acc < 50: suggestions.append({ "type": "WARNING", "priority": "HIGH", "title": "AI Below 50% Accuracy", "message": f"AI accuracy is {ai_acc}%. Market conditions may have changed. Try switching AI mode or increasing confidence threshold.", "action": "Review AI model configuration", }) # Holding time intraday_losses = [r for r in records if r.style == "INTRADAY" and (r.pnl or 0) < 0] if len(intraday_losses) > 3: avg_hold = sum(r.holding_minutes or 0 for r in intraday_losses) / len(intraday_losses) if avg_hold > 180: # held losing intraday > 3 hours suggestions.append({ "type": "WARNING", "priority": "MEDIUM", "title": "Holding Losing INTRADAY Trades Too Long", "message": f"Losing intraday trades held avg {avg_hold:.0f} mins. Tighten INTRADAY SL to 1% to exit faster.", "action": "Reduce INTRADAY stop loss distance", }) if not suggestions: suggestions.append({ "type": "INFO", "priority": "LOW", "title": "Performance looks healthy", "message": f"Win rate: {win_rate}%, R:R: {rr}. Continue monitoring.", }) return suggestions def _empty_stats(self) -> dict: return { "total_trades": 0, "wins": 0, "losses": 0, "win_rate": 0, "total_pnl": 0, "avg_pnl": 0, "avg_win": 0, "avg_loss": 0, "risk_reward": 0, "max_drawdown": 0, "profit_factor": 0, } # Singleton learning_service = LearningService()