""" Phase 5: ML Pattern Recognition and Clustering Machine learning-based trade pattern analysis and clustering """ from fastapi import APIRouter, Query, HTTPException from typing import List, Optional, Dict, Any from datetime import datetime, timedelta from dataclasses import dataclass import random router = APIRouter(prefix="/api/ml-patterns", tags=["ML Pattern Recognition"]) @dataclass class TradeCluster: """Represents a cluster of similar trades""" cluster_id: int name: str size: int avg_win_rate: float avg_profit: float confidence: float characteristics: Dict[str, Any] # Mock ML model results SAMPLE_CLUSTERS = [ { "cluster_id": 1, "name": "Morning Golden Cross Strategy", "size": 12, "avg_win_rate": 72.5, "avg_profit": 245.50, "confidence": 0.89, "characteristics": { "entry_condition": "EMA(12) crosses above EMA(26)", "exit_condition": "RSI > 70 or price closes below EMA(12)", "best_timeframe": "15m", "best_hour": "09:00-11:00", "avg_hold_time": "45 minutes", "risk_reward_ratio": 1.8, }, }, { "cluster_id": 2, "name": "Bollinger Band Breakout", "size": 8, "avg_win_rate": 65.0, "avg_profit": 180.25, "confidence": 0.76, "characteristics": { "entry_condition": "Price breaks above BB Upper band", "exit_condition": "Close inside BB or move stops to breakeven", "best_timeframe": "5m", "best_hour": "10:00-15:00", "avg_hold_time": "30 minutes", "risk_reward_ratio": 1.5, }, }, { "cluster_id": 3, "name": "RSI Oversold Bounce", "size": 15, "avg_win_rate": 58.0, "avg_profit": 120.75, "confidence": 0.71, "characteristics": { "entry_condition": "RSI < 30 + price bounces off support", "exit_condition": "RSI > 70 or initial stop loss", "best_timeframe": "15m", "best_hour": "All hours", "avg_hold_time": "60 minutes", "risk_reward_ratio": 1.3, }, }, { "cluster_id": 4, "name": "MACD Divergence Setup", "size": 6, "avg_win_rate": 83.0, "avg_profit": 320.50, "confidence": 0.92, "characteristics": { "entry_condition": "Price lower high but MACD higher high (bullish)", "exit_condition": "MACD crosses below signal line", "best_timeframe": "1h", "best_hour": "09:00-17:00", "avg_hold_time": "2-4 hours", "risk_reward_ratio": 2.5, }, }, { "cluster_id": 5, "name": "Support Bounce Pattern", "size": 20, "avg_win_rate": 62.0, "avg_profit": 95.30, "confidence": 0.68, "characteristics": { "entry_condition": "Price touches pivot point or key support", "exit_condition": "Next resistance or predetermined TP", "best_timeframe": "5m-15m", "best_hour": "09:00-16:00", "avg_hold_time": "20-45 minutes", "risk_reward_ratio": 1.2, }, }, ] # Mock market condition analysis MARKET_CONDITIONS = { "trending_up": { "name": "Strong Uptrend", "description": "Market in clear uptrend with higher highs and higher lows", "best_clusters": [1, 4], "confidence": 0.87, "recommendation": "Trade breakouts and continuations, avoid shorting", }, "trending_down": { "name": "Strong Downtrend", "description": "Market in clear downtrend with lower highs and lower lows", "best_clusters": [3, 5], "confidence": 0.84, "recommendation": "Trade support bounces, avoid breakout trades", }, "ranging": { "name": "Range-Bound Market", "description": "Market oscillating between support and resistance", "best_clusters": [2, 3, 5], "confidence": 0.72, "recommendation": "Trade bounces off support/resistance, avoid breakouts", }, "volatile": { "name": "High Volatility", "description": "Large price swings with low predictability", "best_clusters": [2, 4], "confidence": 0.65, "recommendation": "Use wider stops, trade divergences, avoid scalping", }, } @router.get("/clusters") async def get_trade_clusters( min_size: int = Query(5, ge=1), min_confidence: float = Query(0.6, ge=0, le=1), sort_by: str = Query("win_rate", regex="^(win_rate|profit|confidence|size)$"), ): """ Get ML-discovered trade clusters - **min_size**: Minimum trades in cluster - **min_confidence**: Minimum confidence score (0-1) - **sort_by**: Sort by win_rate, profit, confidence, or size """ filtered = [c for c in SAMPLE_CLUSTERS if c["size"] >= min_size and c["confidence"] >= min_confidence] # Sort results sort_key = { "win_rate": lambda x: x["avg_win_rate"], "profit": lambda x: x["avg_profit"], "confidence": lambda x: x["confidence"], "size": lambda x: x["size"], }[sort_by] filtered.sort(key=sort_key, reverse=True) return { "total_clusters": len(filtered), "filters_applied": { "min_size": min_size, "min_confidence": min_confidence, }, "clusters": filtered, } @router.get("/cluster/{cluster_id}") async def get_cluster_details(cluster_id: int): """Get detailed analysis of a specific cluster""" cluster = next((c for c in SAMPLE_CLUSTERS if c["cluster_id"] == cluster_id), None) if not cluster: raise HTTPException(status_code=404, detail=f"Cluster {cluster_id} not found") return { "cluster": cluster, "extended_analysis": { "profitability_score": cluster["avg_win_rate"] * cluster["confidence"], "expected_value": ( cluster["avg_profit"] * cluster["avg_win_rate"] / 100 - cluster["avg_profit"] * (1 - cluster["avg_win_rate"] / 100) * 0.7 ), "consistency": f"{cluster['avg_win_rate']:.1f}% of trades profitable", "risk_level": "Low" if cluster["avg_win_rate"] > 70 else "Medium" if cluster["avg_win_rate"] > 55 else "High", "recommended_for": "Aggressive traders" if cluster["avg_profit"] > 200 else "Conservative traders", }, "similar_clusters": [c for c in SAMPLE_CLUSTERS if c["cluster_id"] != cluster_id][:3], } @router.post("/cluster/{cluster_id}/simulate") async def simulate_cluster_trades( cluster_id: int, num_trades: int = Query(100, ge=10, le=1000) ): """Simulate future trades based on cluster characteristics""" cluster = next((c for c in SAMPLE_CLUSTERS if c["cluster_id"] == cluster_id), None) if not cluster: raise HTTPException(status_code=404, detail=f"Cluster {cluster_id} not found") # Simulate trades win_rate = cluster["avg_win_rate"] / 100 simulated_trades = [] cumulative_pnl = 0 for i in range(num_trades): is_win = random.random() < win_rate profit = ( cluster["avg_profit"] * random.uniform(0.7, 1.3) if is_win else -cluster["avg_profit"] * 0.7 * random.uniform(0.7, 1.3) ) cumulative_pnl += profit simulated_trades.append( { "trade_num": i + 1, "result": "Win" if is_win else "Loss", "profit": round(profit, 2), "cumulative_pnl": round(cumulative_pnl, 2), } ) wins = sum(1 for t in simulated_trades if t["result"] == "Win") total_profit = sum(t["profit"] for t in simulated_trades) return { "cluster_id": cluster_id, "simulation_size": num_trades, "simulated_win_rate": f"{wins/num_trades*100:.1f}%", "simulated_total_profit": round(total_profit, 2), "simulated_avg_trade": round(total_profit / num_trades, 2), "best_streak": max((len(list(g)) for k, g in __import__("itertools").groupby(simulated_trades, lambda x: x["result"] == "Win") if k), default=0), "recent_trades": simulated_trades[-10:], } @router.get("/market-condition") async def analyze_market_condition(): """Analyze current market condition and recommend best clusters""" # In production, this would analyze real market data current_condition = "trending_up" condition_data = MARKET_CONDITIONS[current_condition] return { "current_condition": current_condition, "condition_analysis": condition_data, "recommended_clusters": [ SAMPLE_CLUSTERS[SAMPLE_CLUSTERS[0]["cluster_id"] - 1 + i] for i in range(min(len(condition_data["best_clusters"]), 3)) ], "expected_profitability": condition_data["confidence"], "next_update": (datetime.now() + timedelta(minutes=15)).isoformat(), } @router.get("/recommendations") async def get_trading_recommendations( current_price: float = Query(2000.0), timeframe: str = Query("15m", regex="^(1m|5m|15m|1h|4h|1d)$"), ): """Get ML-based trading recommendations""" # Analyze current conditions market_analysis = await analyze_market_condition() recommendations = [] for cluster in SAMPLE_CLUSTERS[:3]: # Top 3 clusters if cluster["best_timeframe"].replace("m", "").replace("h", "") in timeframe: recommendations.append( { "cluster_id": cluster["cluster_id"], "strategy": cluster["name"], "confidence": cluster["confidence"], "win_rate": cluster["avg_win_rate"], "action": "BUY" if market_analysis["current_condition"] == "trending_up" else "SELL", "entry_price": current_price * (1 - 0.002) if "BUY" else current_price * (1 + 0.002), "take_profit": current_price * (1 + cluster["characteristics"]["risk_reward_ratio"] * 0.005), "stop_loss": current_price * (1 - 0.005), "risk_reward": cluster["characteristics"]["risk_reward_ratio"], "probability": round(cluster["avg_win_rate"] * cluster["confidence"], 2), } ) return { "timeframe": timeframe, "current_price": current_price, "market_condition": market_analysis["current_condition"], "recommendations": sorted(recommendations, key=lambda x: x["probability"], reverse=True), "best_recommendation": recommendations[0] if recommendations else None, } @router.get("/similarity/{cluster_id}") async def find_similar_patterns(cluster_id: int): """Find similar trade patterns based on cluster characteristics""" cluster = next((c for c in SAMPLE_CLUSTERS if c["cluster_id"] == cluster_id), None) if not cluster: raise HTTPException(status_code=404, detail=f"Cluster {cluster_id} not found") # Calculate similarity score (simplified) similar = [] for c in SAMPLE_CLUSTERS: if c["cluster_id"] != cluster_id: similarity = ( (1 - abs(c["avg_win_rate"] - cluster["avg_win_rate"]) / 100) + (1 - abs(c["avg_profit"] - cluster["avg_profit"]) / 500) ) / 2 similar.append({"cluster": c, "similarity_score": similarity}) similar.sort(key=lambda x: x["similarity_score"], reverse=True) return { "reference_cluster": cluster, "similar_patterns": [s for s in similar[:5]], "use_case": "Use similar patterns to confirm trade setup validity", } @router.get("/performance-projection") async def project_future_performance( days_ahead: int = Query(30, ge=1, le=90), assumed_trades_per_day: int = Query(5, ge=1, le=50), ): """Project future performance based on ML clusters""" best_cluster = max(SAMPLE_CLUSTERS, key=lambda x: x["avg_win_rate"] * x["confidence"]) total_trades = days_ahead * assumed_trades_per_day win_rate = best_cluster["avg_win_rate"] / 100 wins = int(total_trades * win_rate) losses = total_trades - wins total_profit = wins * best_cluster["avg_profit"] - losses * best_cluster["avg_profit"] * 0.7 return { "projection_period": f"{days_ahead} days", "assumed_trades_per_day": assumed_trades_per_day, "total_projected_trades": total_trades, "projected_wins": wins, "projected_losses": losses, "projected_win_rate": f"{win_rate*100:.1f}%", "projected_total_profit": round(total_profit, 2), "projected_avg_trade_profit": round(total_profit / total_trades, 2), "daily_avg_profit": round(total_profit / days_ahead, 2), "monthly_projection": round(total_profit / days_ahead * 30, 2), "assumptions": [ "Based on best performing cluster", f"Consistent {assumed_trades_per_day} trades per day", "Market conditions remain stable", "No slippage or commissions", ], } @router.post("/feedback/{cluster_id}") async def submit_cluster_feedback( cluster_id: int, actual_win_rate: float = Query(..., ge=0, le=100), feedback: str = Query(...), ): """ Submit feedback on cluster performance for model improvement - **cluster_id**: ID of cluster being evaluated - **actual_win_rate**: Observed win rate in real trading - **feedback**: Qualitative feedback on pattern performance """ cluster = next((c for c in SAMPLE_CLUSTERS if c["cluster_id"] == cluster_id), None) if not cluster: raise HTTPException(status_code=404, detail=f"Cluster {cluster_id} not found") accuracy = abs(cluster["avg_win_rate"] - actual_win_rate) return { "status": "feedback_recorded", "cluster_id": cluster_id, "expected_win_rate": cluster["avg_win_rate"], "actual_win_rate": actual_win_rate, "prediction_accuracy": 100 - accuracy, "feedback": feedback, "message": "Thank you! This feedback helps improve our ML model.", "next_model_update": (datetime.now() + timedelta(days=7)).date().isoformat(), } @router.get("/model-stats") async def get_ml_model_statistics(): """Get statistics about the ML model and its performance""" total_trades_analyzed = sum(c["size"] for c in SAMPLE_CLUSTERS) avg_accuracy = sum(c["confidence"] for c in SAMPLE_CLUSTERS) / len(SAMPLE_CLUSTERS) best_cluster = max(SAMPLE_CLUSTERS, key=lambda x: x["avg_win_rate"] * x["confidence"]) return { "model_info": { "version": "2.1.0", "last_updated": "2024-11-10", "training_data_size": 500, }, "performance": { "clusters_discovered": len(SAMPLE_CLUSTERS), "total_trades_analyzed": total_trades_analyzed, "average_cluster_accuracy": round(avg_accuracy, 3), "best_cluster": best_cluster["name"], "best_cluster_win_rate": f"{best_cluster['avg_win_rate']:.1f}%", }, "ml_algorithms_used": [ "K-Means Clustering", "Feature Extraction (Technical Indicators)", "Win Rate Prediction Model", "Pattern Recognition Neural Network", ], "next_model_retraining": "2024-11-20", }