"""AI Trade Architect — complete AI-generated trade plans. Combines ALL available intelligence sources into a comprehensive trade plan for each asset: - Direction consensus from signals + momentum + ML + mean-reversion - Entry strategy: immediate / limit at support / scale-in at dips - Position sizing: Kelly-adjusted from win probability + risk budget - Stop loss: max of signal SL, adaptive support - ATR, swing low - Take profit: 3 staged TPs from signals, resistance, ML target - Timing: best entry window from momentum + cycle timing - Risk assessment: overall trade risk (1-10) - Conviction score (0-100): how strongly to recommend this trade - Trade rationale: text summary combining all analysis Data sources (all lazy-imported): - Asset : id, symbol, name, icon_thumb_url, asset_type, sector, market_cap_rank, is_active - AssetProfile : current_price_idr, price_change_1h/24h/7d/30d, market_cap_idr, total_volume_idr, ath_idr, atl_idr - TradingSignal : signal_type, confidence, score, entry_price, stop_loss, take_profit_1/2/3, direction_probability, regime, recommended_strategy, safety_rating, mtf_confirmed - BullishMomentumScore : score, ml_trend, ml_short_pct, ml_medium_pct, ml_target_price, momentum_state, velocity_short_pct, velocity_medium_pct, acceleration, rsi_latest, bb_position, zscore, ema_bullish, entry_price, stop_loss, take_profit_1, take_profit_2, upside_pct, safety_rating, confidence, conditions_met, bullish_phase - RangeTradingScore : atr_pct, zscore, bb_width, range_width_pct, nearest_support, nearest_resistance, is_mean_reverting, half_life_candles, adaptive_support, adaptive_resistance, adaptive_range_pct, oscillation_score, bounce_frequency, cycle_success_rate_pct, cycle_confidence, est_profit_per_cycle_pct, mr_confidence """ from __future__ import annotations import logging import math from collections import defaultdict logger = logging.getLogger(__name__) # ── Helpers ───────────────────────────────────────────────────────────── def _safe_float(val, default: float = 0.0) -> float: """Safely convert DB Numeric / Decimal to float.""" if val is None: return default try: f = float(val) return default if (math.isnan(f) or math.isinf(f)) else f except (ValueError, TypeError): return default def _clamp(value: float, lo: float = 0.0, hi: float = 100.0) -> float: return max(lo, min(hi, value)) def _pct_diff(a: float, b: float) -> float: """Percentage difference: (a - b) / b * 100.""" if b <= 0: return 0.0 return (a - b) / b * 100.0 # ── Conviction tiers ──────────────────────────────────────────────────── def _conviction_level(score: float) -> str: if score >= 90: return 'maximum' if score >= 70: return 'high' if score >= 50: return 'moderate' return 'low' # ======================================================================== # Main service # ======================================================================== class AITradeArchitectService: """AI-generated complete trade plans combining all intelligence.""" # Direction consensus weights W_SIGNAL = 0.30 W_MOMENTUM = 0.25 W_ML = 0.25 W_MEAN_REV = 0.20 # Default risk budget per trade DEFAULT_RISK_PCT = 2.0 # ------------------------------------------------------------------ # # Public API # # ------------------------------------------------------------------ # def scan_all( self, asset_type: str = 'crypto', limit: int = 50, page: int = 1, sort_by: str = 'conviction_desc', min_conviction: int = 0, ) -> dict: """Return paginated AI trade plans for all analysable assets. Parameters ---------- asset_type : str 'crypto', 'stock', 'stock_us', 'all' limit : int Items per page. page : int 1-based page number. sort_by : str 'conviction_desc' | 'conviction_asc' | 'expected_return_desc' | 'risk_asc' | 'market_cap' min_conviction : int Only include plans with conviction >= this (0-100). Returns ------- dict with: items, total, page, limit, has_more, stats. """ try: # -- Load assets -- coins_map = self._load_coins(asset_type) if not coins_map: return self._empty_result(page, limit) asset_ids = list(coins_map.keys()) profiles_map = self._load_latest_profiles(asset_ids) signals_map = self._load_latest_signals(asset_ids) bullish_map = self._load_bullish_scores(asset_ids) range_map = self._load_range_scores(asset_ids) # -- Build trade plans -- all_items: list[dict] = [] for asset_id, asset in coins_map.items(): profile = profiles_map.get(asset_id) if not profile: continue price = _safe_float(profile.current_price_idr) if price <= 0: continue signal = signals_map.get(asset_id) bs = bullish_map.get(asset_id) rs = range_map.get(asset_id) # Need at least some data to build a plan if not signal and not bs and not rs: continue item = self._build_trade_plan(asset, profile, signal, bs, rs) if item: all_items.append(item) # -- Filter by min conviction -- if min_conviction > 0: all_items = [ i for i in all_items if i['conviction_score'] >= min_conviction ] # -- Sort -- all_items = self._sort_items(all_items, sort_by) # -- Stats (before pagination) -- stats = self._build_stats(all_items) # -- Paginate -- total = len(all_items) offset = (page - 1) * limit page_items = all_items[offset: offset + limit] has_more = (offset + limit) < total return { 'items': page_items, 'total': total, 'page': page, 'limit': limit, 'has_more': has_more, 'stats': stats, } except Exception as exc: logger.exception('AITradeArchitectService.scan_all failed') return { 'items': [], 'total': 0, 'page': page, 'limit': limit, 'has_more': False, 'stats': {}, 'error': str(exc), } # ------------------------------------------------------------------ # # Trade plan builder # # ------------------------------------------------------------------ # def _build_trade_plan( self, asset, profile, signal, bullish, range_score, ) -> dict | None: """Build a comprehensive trade plan for one asset.""" price = _safe_float(profile.current_price_idr) if price <= 0: return None # ── Extract all available data ────────────────────────── pct_1h = _safe_float(profile.price_change_1h) pct_24h = _safe_float(profile.price_change_24h) pct_7d = _safe_float(profile.price_change_7d) pct_30d = _safe_float(profile.price_change_30d) mcap = _safe_float(profile.market_cap_idr) volume = _safe_float(profile.total_volume_idr) ath = _safe_float(profile.ath_idr) atl = _safe_float(profile.atl_idr) # Signal data sig_type = signal.signal_type if signal else None sig_confidence = signal.confidence if signal else None sig_score = _safe_float(signal.score, 50.0) if signal else 50.0 sig_entry = _safe_float(signal.entry_price) if signal else 0.0 sig_sl = _safe_float(signal.stop_loss) if signal else 0.0 sig_tp1 = _safe_float(signal.take_profit_1) if signal else 0.0 sig_tp2 = _safe_float(signal.take_profit_2) if signal else 0.0 sig_tp3 = _safe_float(signal.take_profit_3) if signal else 0.0 dir_prob = _safe_float(signal.direction_probability, 0.5) if signal else 0.5 regime = (signal.regime or 'unknown') if signal else 'unknown' strategy = (signal.recommended_strategy or 'swing') if signal else 'swing' sig_safety = (signal.safety_rating or 'MODERATE') if signal else 'MODERATE' mtf_confirmed = bool(signal.mtf_confirmed) if signal else False # Bullish momentum data bull_score = _safe_float(bullish.score, 50.0) if bullish else 50.0 ml_trend = (bullish.ml_trend or 'unknown') if bullish else 'unknown' ml_short = _safe_float(bullish.ml_short_pct) if bullish else 0.0 ml_medium = _safe_float(bullish.ml_medium_pct) if bullish else 0.0 ml_target = _safe_float(bullish.ml_target_price) if bullish else 0.0 momentum_state = (bullish.momentum_state or 'STABLE') if bullish else 'STABLE' velocity_short = _safe_float(bullish.velocity_short_pct) if bullish else 0.0 velocity_med = _safe_float(bullish.velocity_medium_pct) if bullish else 0.0 accel = _safe_float(bullish.acceleration) if bullish else 0.0 rsi = _safe_float(bullish.rsi_latest, 50.0) if bullish else 50.0 bb_pos = _safe_float(bullish.bb_position, 0.5) if bullish else 0.5 bull_zscore = _safe_float(bullish.zscore) if bullish else 0.0 ema_bullish = bool(bullish.ema_bullish) if bullish else False bull_entry = _safe_float(bullish.entry_price) if bullish else 0.0 bull_sl = _safe_float(bullish.stop_loss) if bullish else 0.0 bull_tp1 = _safe_float(bullish.take_profit_1) if bullish else 0.0 bull_tp2 = _safe_float(bullish.take_profit_2) if bullish else 0.0 upside_pct = _safe_float(bullish.upside_pct) if bullish else 0.0 bull_safety = (bullish.safety_rating or 'MODERATE') if bullish else 'MODERATE' bull_confidence = _safe_float(bullish.confidence, 0.5) if bullish else 0.5 conditions_met = _safe_float(bullish.conditions_met) if bullish else 0.0 bullish_phase = (bullish.bullish_phase or 'unknown') if bullish else 'unknown' # Range/mean-reversion data atr_pct = _safe_float(range_score.atr_pct, 3.0) if range_score else 3.0 zscore = _safe_float(range_score.zscore) if range_score else 0.0 bb_width = _safe_float(range_score.bb_width, 0.05) if range_score else 0.05 range_width = _safe_float(range_score.range_width_pct, 10.0) if range_score else 10.0 support = _safe_float(range_score.nearest_support) if range_score else 0.0 resistance = _safe_float(range_score.nearest_resistance) if range_score else 0.0 is_mean_rev = bool(range_score.is_mean_reverting) if range_score else False half_life = _safe_float(range_score.half_life_candles) if range_score else 0.0 adaptive_sup = _safe_float(range_score.adaptive_support) if range_score else 0.0 adaptive_res = _safe_float(range_score.adaptive_resistance) if range_score else 0.0 osc_score = _safe_float(range_score.oscillation_score) if range_score else 0.0 bounce_freq = _safe_float(range_score.bounce_frequency) if range_score else 0.0 cycle_success = _safe_float(range_score.cycle_success_rate_pct) if range_score else 0.0 cycle_conf = _safe_float(range_score.cycle_confidence) if range_score else 0.0 est_profit_cycle = _safe_float(range_score.est_profit_per_cycle_pct) if range_score else 0.0 mr_confidence = _safe_float(range_score.mr_confidence) if range_score else 0.0 # ── 1. Direction consensus ────────────────────────────── direction, direction_scores = self._compute_direction( sig_type=sig_type, sig_score=sig_score, dir_prob=dir_prob, bull_score=bull_score, ml_trend=ml_trend, momentum_state=momentum_state, ema_bullish=ema_bullish, is_mean_rev=is_mean_rev, zscore=zscore, rsi=rsi, ) # ── 2. Entry strategy ────────────────────────────────── entry_strategy = self._determine_entry_strategy( direction, price, support, adaptive_sup, zscore, rsi, bb_pos, velocity_short, regime, ) # ── 3. Stop loss ─────────────────────────────────────── sl_price = self._determine_stop_loss( direction, price, sig_sl, bull_sl, adaptive_sup, support, atr_pct, ) # ── 4. Take profit levels ────────────────────────────── tp1, tp2, tp3 = self._determine_take_profits( direction, price, sig_tp1, sig_tp2, sig_tp3, bull_tp1, bull_tp2, ml_target, resistance, adaptive_res, ath, ) # ── 5. Position sizing (Kelly) ───────────────────────── win_prob = max(0.3, min(0.85, dir_prob)) position_size_pct, risk_per_trade = self._kelly_position_size( win_prob, price, sl_price, tp1, ) # ── 6. Timing ────────────────────────────────────────── timing = self._assess_timing( momentum_state, velocity_short, accel, rsi, half_life, bounce_freq, bullish_phase, ) # ── 7. Risk assessment (1-10) ────────────────────────── trade_risk = self._assess_risk( atr_pct, bb_width, zscore, regime, sig_safety, bull_safety, price, sl_price, mcap, ) # ── 8. Conviction score (0-100) ──────────────────────── conviction = self._compute_conviction( direction_scores=direction_scores, sig_confidence=sig_confidence, bull_confidence=bull_confidence, mtf_confirmed=mtf_confirmed, conditions_met=conditions_met, cycle_conf=cycle_conf, mr_confidence=mr_confidence, trade_risk=trade_risk, upside_pct=upside_pct, ) # ── 9. Expected return / duration ────────────────────── if direction == 'BUY' and tp1 > price: expected_return = _pct_diff(tp1, price) elif direction == 'SELL' and tp1 > 0 and tp1 < price: expected_return = _pct_diff(price, tp1) else: expected_return = 0.0 expected_duration = self._estimate_duration( strategy, half_life, atr_pct, ) # ── 10. Trade rationale ──────────────────────────────── rationale = self._build_rationale( asset, direction, entry_strategy, conviction, trade_risk, sig_type, ml_trend, is_mean_rev, zscore, rsi, momentum_state, expected_return, tp1, sl_price, price, ) conv_level = _conviction_level(conviction) return { 'asset_id': asset.id, 'symbol': asset.symbol, 'name': asset.name, 'icon_url': asset.icon_thumb_url or asset.image_url or '', 'asset_type': asset.asset_type or 'crypto', 'sector': asset.sector or 'Unknown', 'market_cap_rank': asset.market_cap_rank or 9999, 'current_price_idr': round(price, 8), 'price_change_1h': round(pct_1h, 2), 'price_change_24h': round(pct_24h, 2), 'price_change_7d': round(pct_7d, 2), 'price_change_30d': round(pct_30d, 2), 'market_cap_idr': round(mcap, 0), 'volume_idr': round(volume, 0), # -- Trade plan core -- 'direction': direction, 'direction_scores': direction_scores, 'entry_type': entry_strategy['type'], 'entry_price': round(entry_strategy['price'], 8), 'entry_reasoning': entry_strategy['reasoning'], 'sl_price': round(sl_price, 8), 'tp1': round(tp1, 8), 'tp2': round(tp2, 8), 'tp3': round(tp3, 8), 'position_size_pct': round(position_size_pct, 2), 'risk_per_trade_pct': round(risk_per_trade, 2), 'expected_return_pct': round(expected_return, 2), 'expected_duration_days': expected_duration, # -- Scores -- 'conviction_score': round(conviction, 1), 'conviction_level': conv_level, 'trade_risk': trade_risk, 'timing': timing, # -- Supporting intelligence -- 'signal_type': sig_type, 'signal_score': round(sig_score, 1), 'signal_confidence': sig_confidence, 'regime': regime, 'strategy': strategy, 'mtf_confirmed': mtf_confirmed, 'momentum_score': round(bull_score, 1), 'ml_trend': ml_trend, 'momentum_state': momentum_state, 'rsi': round(rsi, 1), 'zscore': round(zscore, 3), 'atr_pct': round(atr_pct, 4), 'is_mean_reverting': is_mean_rev, 'safety_rating': sig_safety, 'upside_pct': round(upside_pct, 2), # -- Rationale -- 'trade_rationale': rationale, } # ------------------------------------------------------------------ # # Direction consensus # # ------------------------------------------------------------------ # def _compute_direction( self, sig_type, sig_score: float, dir_prob: float, bull_score: float, ml_trend: str, momentum_state: str, ema_bullish: bool, is_mean_rev: bool, zscore: float, rsi: float, ) -> tuple[str, dict]: """Compute direction consensus from all sources. Returns (direction, direction_scores_dict). """ scores = {} # 1. Signal vote (BUY=+1, SELL=-1, HOLD=0) if sig_type == 'BUY': sig_vote = min(1.0, sig_score / 80.0) elif sig_type == 'SELL': sig_vote = max(-1.0, -sig_score / 80.0) else: sig_vote = 0.0 scores['signal'] = round(sig_vote, 3) # 2. Momentum vote if bull_score > 60: mom_vote = min(1.0, (bull_score - 50) / 40.0) elif bull_score < 40: mom_vote = max(-1.0, (bull_score - 50) / 40.0) else: mom_vote = 0.0 # Momentum state modifier state_mods = { 'OVERBOUGHT': -0.3, 'OVERSOLD': 0.3, 'ACCELERATING': 0.2, 'DECELERATING': -0.2, } mom_vote += state_mods.get(momentum_state, 0.0) mom_vote = max(-1.0, min(1.0, mom_vote)) scores['momentum'] = round(mom_vote, 3) # 3. ML vote ml_votes = {'bullish': 0.7, 'bearish': -0.7, 'neutral': 0.0, 'unknown': 0.0} ml_vote = ml_votes.get(ml_trend, 0.0) if ema_bullish: ml_vote += 0.2 ml_vote = max(-1.0, min(1.0, ml_vote)) scores['ml'] = round(ml_vote, 3) # 4. Mean-reversion vote if is_mean_rev: if zscore < -1.5: mr_vote = min(1.0, abs(zscore) / 3.0) # Oversold -> BUY elif zscore > 1.5: mr_vote = max(-1.0, -abs(zscore) / 3.0) # Overbought -> SELL else: mr_vote = 0.0 else: mr_vote = 0.0 # RSI extreme modifier if rsi < 30: mr_vote += 0.3 elif rsi > 70: mr_vote -= 0.3 mr_vote = max(-1.0, min(1.0, mr_vote)) scores['mean_reversion'] = round(mr_vote, 3) # Weighted consensus consensus = ( self.W_SIGNAL * sig_vote + self.W_MOMENTUM * mom_vote + self.W_ML * ml_vote + self.W_MEAN_REV * mr_vote ) if consensus > 0.15: direction = 'BUY' elif consensus < -0.15: direction = 'SELL' else: direction = 'NEUTRAL' scores['consensus'] = round(consensus, 3) return direction, scores # ------------------------------------------------------------------ # # Entry strategy # # ------------------------------------------------------------------ # def _determine_entry_strategy( self, direction: str, price: float, support: float, adaptive_sup: float, zscore: float, rsi: float, bb_pos: float, velocity: float, regime: str, ) -> dict: """Determine the best entry strategy.""" if direction == 'NEUTRAL': return { 'type': 'wait', 'price': price, 'reasoning': 'No clear direction — wait for stronger signal.', } if direction == 'BUY': eff_support = adaptive_sup if adaptive_sup > 0 else support # Immediate entry: strong momentum, RSI not extreme if velocity > 1.0 and rsi < 65 and zscore < 1.0: return { 'type': 'immediate', 'price': price, 'reasoning': ( 'Strong upward momentum with room to run. ' f'RSI at {rsi:.0f}, z-score {zscore:.2f}. Enter now.' ), } # Limit at support: price near support if eff_support > 0 and _pct_diff(price, eff_support) < 5: limit_price = round(eff_support * 1.005, 8) return { 'type': 'limit_at_support', 'price': limit_price, 'reasoning': ( f'Price near support at {eff_support:,.2f}. ' f'Place limit buy just above support for better entry.' ), } # Scale-in at dips dip_entry = round(price * 0.97, 8) return { 'type': 'scale_in', 'price': dip_entry, 'reasoning': ( 'Scale into position on 3% dips. ' 'Split entry into 3 tranches to average price.' ), } # direction == 'SELL' return { 'type': 'immediate' if velocity < -1.0 else 'limit_at_resistance', 'price': price, 'reasoning': ( 'Bearish direction detected. ' 'Exit positions or set limit sell at resistance.' ), } # ------------------------------------------------------------------ # # Stop loss # # ------------------------------------------------------------------ # def _determine_stop_loss( self, direction: str, price: float, sig_sl: float, bull_sl: float, adaptive_sup: float, support: float, atr_pct: float, ) -> float: """Determine optimal stop loss price.""" if direction == 'NEUTRAL': return 0.0 atr_distance = price * max(atr_pct, 0.5) / 100.0 if direction == 'BUY': candidates = [] if sig_sl > 0 and sig_sl < price: candidates.append(sig_sl) if bull_sl > 0 and bull_sl < price: candidates.append(bull_sl) if adaptive_sup > 0 and adaptive_sup < price: candidates.append(adaptive_sup - atr_distance * 0.5) if support > 0 and support < price: candidates.append(support - atr_distance * 0.5) candidates.append(price - atr_distance * 2.0) return max(candidates) if candidates else price * 0.95 else: candidates = [] if sig_sl > 0 and sig_sl > price: candidates.append(sig_sl) candidates.append(price + atr_distance * 2.0) return min(candidates) if candidates else price * 1.05 # ------------------------------------------------------------------ # # Take profit levels # # ------------------------------------------------------------------ # def _determine_take_profits( self, direction: str, price: float, sig_tp1: float, sig_tp2: float, sig_tp3: float, bull_tp1: float, bull_tp2: float, ml_target: float, resistance: float, adaptive_res: float, ath: float, ) -> tuple[float, float, float]: """Determine 3 staged take-profit levels.""" if direction == 'NEUTRAL': return 0.0, 0.0, 0.0 if direction == 'BUY': candidates = [] for tp in [sig_tp1, sig_tp2, sig_tp3, bull_tp1, bull_tp2, ml_target]: if tp > price: candidates.append(tp) if adaptive_res > price: candidates.append(adaptive_res) if resistance > price: candidates.append(resistance) if ath > price: candidates.append(ath) candidates = sorted(set(candidates)) if len(candidates) >= 3: n = len(candidates) tp1 = candidates[0] tp2 = candidates[n // 2] tp3 = candidates[-1] elif len(candidates) == 2: tp1 = candidates[0] tp2 = candidates[1] tp3 = candidates[1] * 1.10 elif len(candidates) == 1: tp1 = candidates[0] tp2 = candidates[0] * 1.10 tp3 = candidates[0] * 1.25 else: tp1 = price * 1.05 tp2 = price * 1.12 tp3 = price * 1.25 return tp1, tp2, tp3 else: return price * 0.95, price * 0.90, price * 0.80 # ------------------------------------------------------------------ # # Position sizing (Kelly criterion) # # ------------------------------------------------------------------ # def _kelly_position_size( self, win_prob: float, price: float, sl_price: float, tp_price: float, ) -> tuple[float, float]: """Calculate Kelly-adjusted position size.""" if sl_price <= 0 or tp_price <= 0 or price <= 0: return 5.0, self.DEFAULT_RISK_PCT risk_pct = abs(_pct_diff(sl_price, price)) reward_pct = abs(_pct_diff(tp_price, price)) if risk_pct <= 0: risk_pct = 3.0 rr_ratio = reward_pct / risk_pct if risk_pct > 0 else 1.0 b = rr_ratio p = win_prob q = 1.0 - p kelly_fraction = (b * p - q) / b if b > 0 else 0.0 half_kelly = max(0.01, kelly_fraction * 0.5) risk_per_trade = min(self.DEFAULT_RISK_PCT, risk_pct * half_kelly) position_size = (risk_per_trade / risk_pct * 100) if risk_pct > 0 else 5.0 position_size = _clamp(position_size, 1.0, 25.0) risk_per_trade = _clamp(risk_per_trade, 0.5, 5.0) return position_size, risk_per_trade # ------------------------------------------------------------------ # # Timing assessment # # ------------------------------------------------------------------ # def _assess_timing( self, momentum_state: str, velocity: float, accel: float, rsi: float, half_life: float, bounce_freq: float, bullish_phase: str, ) -> dict: """Assess optimal entry timing.""" timing_score = 50 if accel > 0.05: timing_score += 15 elif accel < -0.05: timing_score -= 15 if 35 < rsi < 65: timing_score += 10 elif rsi < 25 or rsi > 75: timing_score -= 10 if velocity > 1: timing_score += 10 elif velocity < -1: timing_score -= 5 phase_scores = { 'accumulation': 20, 'markup': 10, 'distribution': -15, 'markdown': -20, 'early_bullish': 15, } timing_score += phase_scores.get(bullish_phase, 0) timing_score = int(_clamp(timing_score)) if timing_score >= 75: quality = 'excellent' window = 'Enter within 24 hours' elif timing_score >= 55: quality = 'good' window = 'Enter within 2-3 days' elif timing_score >= 35: quality = 'fair' window = 'Wait for better setup (3-7 days)' else: quality = 'poor' window = 'Wait for significant pullback or reversal' reasoning = ( f'Timing score {timing_score}/100. ' f'Phase: {bullish_phase}, momentum: {momentum_state}. ' f'RSI: {rsi:.0f}.' ) return { 'quality': quality, 'score': timing_score, 'window': window, 'reasoning': reasoning, } # ------------------------------------------------------------------ # # Risk assessment # # ------------------------------------------------------------------ # def _assess_risk( self, atr_pct: float, bb_width: float, zscore: float, regime: str, sig_safety: str, bull_safety: str, price: float, sl_price: float, mcap: float, ) -> int: """Assess overall trade risk on scale 1-10.""" risk = 5 if atr_pct > 8: risk += 2 elif atr_pct > 5: risk += 1 elif atr_pct < 2: risk -= 1 if bb_width > 0.15: risk += 1 if abs(zscore) > 2.5: risk += 1 if regime in ('volatile', 'trending_down'): risk += 1 elif regime == 'ranging': risk -= 1 safety_risk = {'SAFE': -1, 'MODERATE': 0, 'RISKY': 1, 'DANGEROUS': 2} risk += safety_risk.get(sig_safety, 0) if price > 0 and sl_price > 0: sl_dist = abs(price - sl_price) / price * 100 if sl_dist > 10: risk += 1 elif sl_dist < 3: risk -= 1 if mcap < 1e9: risk += 1 elif mcap > 1e11: risk -= 1 return max(1, min(10, risk)) # ------------------------------------------------------------------ # # Conviction score # # ------------------------------------------------------------------ # def _compute_conviction( self, direction_scores: dict, sig_confidence: str | None, bull_confidence: float, mtf_confirmed: bool, conditions_met: float, cycle_conf: float, mr_confidence: float, trade_risk: int, upside_pct: float, ) -> float: """Compute conviction score 0-100.""" consensus = abs(direction_scores.get('consensus', 0)) # Source agreement (0-40) votes = [ direction_scores.get('signal', 0), direction_scores.get('momentum', 0), direction_scores.get('ml', 0), direction_scores.get('mean_reversion', 0), ] consensus_sign = 1 if direction_scores.get('consensus', 0) >= 0 else -1 agreeing = sum(1 for v in votes if v * consensus_sign > 0.1) agreement_score = agreeing * 10 # Confidence components (0-25) conf_map = {'High': 20, 'Medium': 12, 'Low': 5} conf_score = conf_map.get(sig_confidence, 8) conf_score += bull_confidence * 5 # Confirmations (0-20) confirm_score = 0 if mtf_confirmed: confirm_score += 8 confirm_score += min(8, conditions_met * 2) if cycle_conf > 0.5: confirm_score += 4 # Risk inverse (0-15) risk_score = max(0, (10 - trade_risk)) * 1.5 raw = agreement_score + conf_score + confirm_score + risk_score if upside_pct > 10: raw += 5 elif upside_pct > 20: raw += 10 return _clamp(raw) # ------------------------------------------------------------------ # # Expected duration # # ------------------------------------------------------------------ # def _estimate_duration( self, strategy: str, half_life: float, atr_pct: float, ) -> int: """Estimate trade duration in days.""" strategy_days = { 'scalping': 1, 'short_term': 3, 'swing': 7, 'medium_term': 14, 'long_term': 30, } base = strategy_days.get(strategy, 7) if half_life > 0: hl_days = max(1, int(half_life)) base = (base + hl_days) // 2 if atr_pct > 6: base = max(1, base - 2) return max(1, min(60, base)) # ------------------------------------------------------------------ # # Trade rationale # # ------------------------------------------------------------------ # def _build_rationale( self, asset, direction: str, entry_strategy: dict, conviction: float, trade_risk: int, sig_type, ml_trend: str, is_mean_rev: bool, zscore: float, rsi: float, momentum_state: str, expected_return: float, tp1: float, sl: float, price: float, ) -> str: """Build comprehensive trade rationale text.""" parts = [] if direction == 'BUY': parts.append( f'{asset.symbol}: BUY recommendation with {_conviction_level(conviction)} ' f'conviction ({conviction:.0f}/100).' ) elif direction == 'SELL': parts.append( f'{asset.symbol}: SELL recommendation with {_conviction_level(conviction)} ' f'conviction ({conviction:.0f}/100).' ) else: parts.append( f'{asset.symbol}: NEUTRAL — no clear trade setup. ' f'Conviction {conviction:.0f}/100.' ) return ' '.join(parts) # Source agreement sources = [] if sig_type and sig_type == direction: sources.append('trading signal') if ml_trend == 'bullish' and direction == 'BUY': sources.append('ML model (bullish)') elif ml_trend == 'bearish' and direction == 'SELL': sources.append('ML model (bearish)') if is_mean_rev and ((zscore < -1 and direction == 'BUY') or (zscore > 1 and direction == 'SELL')): sources.append(f'mean-reversion (z={zscore:.1f})') if sources: parts.append(f'Supported by: {", ".join(sources)}.') parts.append(f'RSI: {rsi:.0f}, Momentum: {momentum_state}.') parts.append(f'Entry: {entry_strategy["type"]} at {entry_strategy["price"]:,.2f}.') if tp1 > 0 and sl > 0: sl_dist = abs(_pct_diff(sl, price)) tp_dist = abs(_pct_diff(tp1, price)) rr = tp_dist / sl_dist if sl_dist > 0 else 0 parts.append( f'TP1: {tp1:,.2f} ({_pct_diff(tp1, price):+.1f}%), ' f'SL: {sl:,.2f} ({_pct_diff(sl, price):+.1f}%). ' f'R:R = {rr:.1f}x.' ) risk_labels = { 1: 'very low', 2: 'very low', 3: 'low', 4: 'moderate', 5: 'moderate', 6: 'elevated', 7: 'high', 8: 'high', 9: 'very high', 10: 'extreme', } parts.append(f'Trade risk: {risk_labels.get(trade_risk, "unknown")} ({trade_risk}/10).') return ' '.join(parts) # ------------------------------------------------------------------ # # Sorting # # ------------------------------------------------------------------ # def _sort_items(self, items: list[dict], sort_by: str) -> list[dict]: sort_configs = { 'conviction_desc': (lambda r: r['conviction_score'], True), 'conviction_asc': (lambda r: r['conviction_score'], False), 'expected_return_desc': (lambda r: r['expected_return_pct'], True), 'risk_asc': (lambda r: r['trade_risk'], False), 'market_cap': (lambda r: r['market_cap_rank'], False), } key_fn, reverse = sort_configs.get(sort_by, sort_configs['conviction_desc']) items.sort(key=key_fn, reverse=reverse) return items # ------------------------------------------------------------------ # # Stats # # ------------------------------------------------------------------ # def _build_stats(self, items: list[dict]) -> dict: total = len(items) if total == 0: return { 'total_analysed': 0, 'conviction_distribution': {}, 'direction_distribution': {}, 'avg_conviction': 0.0, 'avg_expected_return': 0.0, 'avg_trade_risk': 0.0, 'best_by_conviction': [], 'best_by_return': [], } conv_dist: dict[str, int] = defaultdict(int) dir_dist: dict[str, int] = defaultdict(int) total_conv = 0.0 total_ret = 0.0 total_risk = 0.0 for item in items: conv_dist[item['conviction_level']] += 1 dir_dist[item['direction']] += 1 total_conv += item['conviction_score'] total_ret += item['expected_return_pct'] total_risk += item['trade_risk'] conv_sorted = sorted(items, key=lambda x: x['conviction_score'], reverse=True) best_conv = [ { 'symbol': i['symbol'], 'direction': i['direction'], 'conviction_score': i['conviction_score'], 'expected_return_pct': i['expected_return_pct'], } for i in conv_sorted[:5] ] ret_sorted = sorted(items, key=lambda x: x['expected_return_pct'], reverse=True) best_ret = [ { 'symbol': i['symbol'], 'direction': i['direction'], 'conviction_score': i['conviction_score'], 'expected_return_pct': i['expected_return_pct'], } for i in ret_sorted[:5] ] return { 'total_analysed': total, 'conviction_distribution': dict(conv_dist), 'direction_distribution': dict(dir_dist), 'avg_conviction': round(total_conv / total, 1), 'avg_expected_return': round(total_ret / total, 2), 'avg_trade_risk': round(total_risk / total, 1), 'best_by_conviction': best_conv, 'best_by_return': best_ret, } # ------------------------------------------------------------------ # # Data loading # # ------------------------------------------------------------------ # def _load_coins(self, asset_type: str) -> dict: from app.models.asset import Asset query = Asset.query.filter(Asset.is_active.is_(True)) if asset_type and asset_type != 'all': query = query.filter(Asset.asset_type == asset_type) return {c.id: c for c in query.all()} def _load_latest_profiles(self, asset_ids: list) -> dict: from app.extensions import db from app.models.asset import AssetProfile from sqlalchemy import func as sa_func if not asset_ids: return {} latest_sq = ( db.session.query( AssetProfile.asset_id, sa_func.max(AssetProfile.id).label('max_id'), ) .filter(AssetProfile.asset_id.in_(asset_ids)) .group_by(AssetProfile.asset_id) .subquery() ) profiles = AssetProfile.query.join( latest_sq, AssetProfile.id == latest_sq.c.max_id, ).all() return {p.asset_id: p for p in profiles} def _load_latest_signals(self, asset_ids: list) -> dict: """Load the latest active signal per asset.""" from app.models.signal import TradingSignal if not asset_ids: return {} signals = ( TradingSignal.query .filter( TradingSignal.asset_id.in_(asset_ids), TradingSignal.status == 'active', ) .order_by(TradingSignal.created_at.desc()) .all() ) result = {} for sig in signals: if sig.asset_id not in result: result[sig.asset_id] = sig return result def _load_bullish_scores(self, asset_ids: list) -> dict: from app.models.bullish_score import BullishMomentumScore if not asset_ids: return {} rows = BullishMomentumScore.query.filter( BullishMomentumScore.asset_id.in_(asset_ids), ).all() return {b.asset_id: b for b in rows} def _load_range_scores(self, asset_ids: list) -> dict: from app.models.range_score import RangeTradingScore if not asset_ids: return {} rows = RangeTradingScore.query.filter( RangeTradingScore.asset_id.in_(asset_ids), ).all() return {r.asset_id: r for r in rows} # ------------------------------------------------------------------ # # Empty result # # ------------------------------------------------------------------ # @staticmethod def _empty_result(page: int = 1, limit: int = 50) -> dict: return { 'items': [], 'total': 0, 'page': page, 'limit': limit, 'has_more': False, 'stats': { 'total_analysed': 0, 'conviction_distribution': {}, 'direction_distribution': {}, 'avg_conviction': 0.0, 'avg_expected_return': 0.0, 'avg_trade_risk': 0.0, 'best_by_conviction': [], 'best_by_return': [], }, }