""" Position Architect Service --------------------------- Build position entry plans with: - Entry zone (support -> 10% above) - 4 scaling levels - Stop loss from signal/bullish data - TP ladder (TP1, TP2, resistance) - Risk:Reward ratio - Position size suggestion Signal filter: all / active_buy / high_confidence. """ import math import logging from datetime import datetime logger = logging.getLogger(__name__) def _safe_float(val, default=0.0): """Convert value to float safely, handling None/NaN/Inf.""" if val is None: return default try: f = float(val) if math.isnan(f) or math.isinf(f): return default return f except (ValueError, TypeError): return default def _safe_int(val, default=0): if val is None: return default try: return int(val) except (ValueError, TypeError): return default def _clamp(val, lo=0.0, hi=100.0): return max(lo, min(hi, val)) class PositionArchitectService: """ Design optimal position entry plans: entry zones, scaling levels, stop-losses, take-profit ladders, and risk-reward analysis. """ SIGNAL_FILTER_OPTIONS = ('all', 'active_buy', 'high_confidence') # Default position sizing parameters DEFAULT_CAPITAL = 10_000_000 # 10M IDR MAX_RISK_PCT = 2.0 # max 2% risk per position SCALING_RATIOS = [0.40, 0.25, 0.20, 0.15] # 4 scaling levels SORT_OPTIONS = { 'opportunity_desc': ('opportunity_score', True), 'opportunity_asc': ('opportunity_score', False), 'rr_ratio_desc': ('risk_reward_ratio', True), 'rr_ratio_asc': ('risk_reward_ratio', False), 'upside_desc': ('upside_pct', True), 'confidence_desc': ('signal_confidence', True), 'name_asc': ('symbol', False), 'name_desc': ('symbol', True), } # ----------------------------------------------------------------------- # Public API # ----------------------------------------------------------------------- def scan_all( self, asset_type='crypto', limit=50, page=1, sort_by='opportunity_desc', signal_status='all', ): """ Build position plans for qualifying assets. Parameters ---------- asset_type : str limit : int page : int sort_by : str signal_status : str – all / active_buy / high_confidence Returns ------- dict """ try: limit = max(1, min(200, _safe_int(limit, 50))) page = max(1, _safe_int(page, 1)) if signal_status not in self.SIGNAL_FILTER_OPTIONS: signal_status = 'all' raw = self._fetch_data(asset_type) if not raw: return self._empty_result(page, limit) scored = [] for row in raw: item = self._build_position_plan(row) if item is None: continue if not self._passes_signal_filter(item, signal_status): continue scored.append(item) if not scored: return self._empty_result(page, limit) scored = self._sort_items(scored, sort_by) total = len(scored) start = (page - 1) * limit end = start + limit page_items = scored[start:end] has_more = end < total stats = self._compute_stats(scored) return { 'items': page_items, 'total': total, 'page': page, 'limit': limit, 'has_more': has_more, 'stats': stats, } except Exception as exc: logger.exception("PositionArchitectService.scan_all failed: %s", exc) return self._empty_result(page, limit) # ----------------------------------------------------------------------- # Data fetching # ----------------------------------------------------------------------- def _fetch_data(self, asset_type): from app.extensions import db from app.models.asset import Asset, AssetProfile from app.models.range_score import RangeTradingScore from app.models.bullish_score import BullishMomentumScore from app.models.signal import TradingSignal from sqlalchemy import func try: latest_profile = ( db.session.query( AssetProfile.asset_id, func.max(AssetProfile.id).label('max_id'), ) .group_by(AssetProfile.asset_id) .subquery() ) latest_range = ( db.session.query( RangeTradingScore.asset_id, func.max(RangeTradingScore.id).label('max_id'), ) .group_by(RangeTradingScore.asset_id) .subquery() ) latest_bullish = ( db.session.query( BullishMomentumScore.asset_id, func.max(BullishMomentumScore.id).label('max_id'), ) .group_by(BullishMomentumScore.asset_id) .subquery() ) latest_signal = ( db.session.query( TradingSignal.asset_id, func.max(TradingSignal.id).label('max_id'), ) .group_by(TradingSignal.asset_id) .subquery() ) query = ( db.session.query( Asset, AssetProfile, RangeTradingScore, BullishMomentumScore, TradingSignal, ) .filter(Asset.is_active.is_(True)) .filter(Asset.asset_type == asset_type) .outerjoin(latest_profile, latest_profile.c.asset_id == Asset.id) .outerjoin(AssetProfile, AssetProfile.id == latest_profile.c.max_id) .outerjoin(latest_range, latest_range.c.asset_id == Asset.id) .outerjoin(RangeTradingScore, RangeTradingScore.id == latest_range.c.max_id) .outerjoin(latest_bullish, latest_bullish.c.asset_id == Asset.id) .outerjoin(BullishMomentumScore, BullishMomentumScore.id == latest_bullish.c.max_id) .outerjoin(latest_signal, latest_signal.c.asset_id == Asset.id) .outerjoin(TradingSignal, TradingSignal.id == latest_signal.c.max_id) ) return query.all() except Exception as exc: logger.error("_fetch_data error: %s", exc) return [] # ----------------------------------------------------------------------- # Position plan building # ----------------------------------------------------------------------- def _build_position_plan(self, row): try: asset, profile, rts, bullish, signal = row if asset is None: return None symbol = getattr(asset, 'symbol', '') or '' name = getattr(asset, 'name', '') or '' sector = getattr(asset, 'sector', '') or 'unknown' icon = getattr(asset, 'icon_thumb_url', '') or '' image = getattr(asset, 'image_url', '') or '' market_cap_rank = _safe_int(getattr(asset, 'market_cap_rank', None), 9999) current_price = _safe_float(getattr(profile, 'current_price_idr', None)) ath = _safe_float(getattr(profile, 'ath_idr', None)) market_cap = _safe_float(getattr(profile, 'market_cap_idr', None)) volume_24h = _safe_float(getattr(profile, 'total_volume_idr', None)) change_24h = _safe_float(getattr(profile, 'price_change_24h', None)) change_7d = _safe_float(getattr(profile, 'price_change_7d', None)) support = _safe_float(getattr(rts, 'nearest_support', None)) resistance = _safe_float(getattr(rts, 'nearest_resistance', None)) range_width = _safe_float(getattr(rts, 'range_width_pct', None)) zscore = _safe_float(getattr(rts, 'zscore', None)) is_mr = bool(getattr(rts, 'is_mean_reverting', False)) mr_conf = _safe_float(getattr(rts, 'mr_confidence', None)) cycle_sr = _safe_float(getattr(rts, 'cycle_success_rate_pct', None)) _rts_price = _safe_float(getattr(rts, 'current_price_idr', None)) range_pos = ((_rts_price - support) / (resistance - support) * 100) if resistance > support > 0 and _rts_price > 0 else None bull_score = _safe_float(getattr(bullish, 'score', None)) phase = getattr(bullish, 'bullish_phase', None) or 'NOT_BULLISH' mom_state = getattr(bullish, 'momentum_state', None) or 'STABLE' vel_short = _safe_float(getattr(bullish, 'velocity_short_pct', None)) vel_med = _safe_float(getattr(bullish, 'velocity_medium_pct', None)) accel = _safe_float(getattr(bullish, 'acceleration', None)) ml_trend = getattr(bullish, 'ml_trend', None) or '' ml_target = _safe_float(getattr(bullish, 'ml_target_price', None)) upside = _safe_float(getattr(bullish, 'upside_pct', None)) tp1 = _safe_float(getattr(bullish, 'take_profit_1', None)) tp2 = _safe_float(getattr(bullish, 'take_profit_2', None)) b_sl = _safe_float(getattr(bullish, 'stop_loss', None)) sig_type = getattr(signal, 'signal_type', None) or '' sig_status = getattr(signal, 'status', None) or '' dir_prob = _safe_float(getattr(signal, 'direction_probability', None)) sig_conf = _safe_float(getattr(signal, 'confidence', None)) regime = getattr(signal, 'regime', None) or '' mtf = bool(getattr(signal, 'mtf_confirmed', False)) sig_entry = _safe_float(getattr(signal, 'entry_price', None)) sig_target = _safe_float(getattr(signal, 'take_profit_1', None)) sig_sl = _safe_float(getattr(signal, 'stop_loss', None)) if current_price <= 0: return None # ---- 1. Entry Zone ---- entry_zone = self._calc_entry_zone(current_price, support, sig_entry) # ---- 2. Scaling Levels ---- scaling_levels = self._calc_scaling_levels(entry_zone) # ---- 3. Stop Loss ---- stop_loss = self._determine_stop_loss( b_sl, sig_sl, support, current_price, ) # ---- 4. Take Profit Ladder ---- tp_ladder = self._calc_tp_ladder( tp1, tp2, resistance, sig_target, ml_target, current_price, ath, ) # ---- 5. Risk:Reward Ratio ---- avg_entry = (entry_zone['low'] + entry_zone['high']) / 2.0 risk_reward = self._calc_risk_reward( avg_entry, stop_loss, tp_ladder, ) # ---- 6. Position Size Suggestion ---- position_sizing = self._calc_position_size( avg_entry, stop_loss, sig_conf, bull_score, ) # ---- 7. Opportunity Score (0-100) ---- opportunity_score = self._calc_opportunity_score( risk_reward, sig_conf, dir_prob, bull_score, phase, mom_state, mtf, upside, range_pos, zscore, mr_conf, ) # Grade if opportunity_score >= 80: grade = 'A' elif opportunity_score >= 65: grade = 'B' elif opportunity_score >= 50: grade = 'C' elif opportunity_score >= 35: grade = 'D' else: grade = 'F' # Position type recommendation position_type = self._recommend_position_type( phase, range_pos, zscore, sig_type, range_width, is_mr, ) return { 'asset_id': asset.id, 'symbol': symbol.upper(), 'name': name, 'sector': sector, 'icon_thumb_url': icon, 'image_url': image, 'market_cap_rank': market_cap_rank, 'current_price_idr': current_price, 'market_cap_idr': market_cap, 'volume_24h_idr': volume_24h, 'price_change_24h': change_24h, 'price_change_7d': change_7d, 'opportunity_score': opportunity_score, 'grade': grade, 'position_type': position_type, 'entry_zone': entry_zone, 'scaling_levels': scaling_levels, 'stop_loss_price': stop_loss, 'stop_loss_pct': round( ((avg_entry - stop_loss) / avg_entry * 100) if avg_entry > 0 and stop_loss > 0 else 0, 2 ), 'tp_ladder': tp_ladder, 'risk_reward_ratio': risk_reward.get('ratio', 0), 'risk_reward_detail': risk_reward, 'position_sizing': position_sizing, 'upside_pct': upside, 'bullish_phase': phase, 'momentum_state': mom_state, 'bullish_score': bull_score, 'ml_trend': ml_trend, 'ml_target_price': ml_target, 'velocity_short_pct': vel_short, 'velocity_medium_pct': vel_med, 'acceleration': accel, 'zscore': zscore, 'range_position_pct': range_pos, 'range_width_pct': range_width, 'support_price': support, 'resistance_price': resistance, 'signal_type': sig_type, 'signal_status': sig_status, 'signal_confidence': sig_conf, 'direction_probability': dir_prob, 'regime': regime, 'mtf_confirmed': mtf, } except Exception as exc: logger.warning("_build_position_plan error: %s", exc) return None # ----------------------------------------------------------------------- # Entry zone # ----------------------------------------------------------------------- def _calc_entry_zone(self, price, support, sig_entry): """Calculate entry zone from support to 10% above support.""" if support > 0: zone_low = support zone_high = support * 1.10 elif sig_entry > 0: zone_low = sig_entry * 0.97 zone_high = sig_entry * 1.05 else: zone_low = price * 0.95 zone_high = price * 1.02 # Ensure zone makes sense relative to current price if zone_high < price * 0.80: zone_low = price * 0.95 zone_high = price * 1.02 return { 'low': round(zone_low, 2), 'high': round(zone_high, 2), 'mid': round((zone_low + zone_high) / 2.0, 2), 'width_pct': round( ((zone_high - zone_low) / zone_low * 100) if zone_low > 0 else 0, 2, ), } # ----------------------------------------------------------------------- # Scaling levels # ----------------------------------------------------------------------- def _calc_scaling_levels(self, entry_zone): """Calculate 4 scaling entry levels within the entry zone.""" low = entry_zone['low'] high = entry_zone['high'] if low <= 0 or high <= 0 or high <= low: return [] step = (high - low) / 3.0 # 4 levels across the zone levels = [] for i, ratio in enumerate(self.SCALING_RATIOS): level_price = low + step * i levels.append({ 'level': i + 1, 'price': round(level_price, 2), 'allocation_pct': round(ratio * 100, 1), 'label': self._scaling_label(i), }) return levels @staticmethod def _scaling_label(index): labels = ['INITIAL', 'SECONDARY', 'TERTIARY', 'FINAL'] return labels[index] if index < len(labels) else f'LEVEL_{index + 1}' # ----------------------------------------------------------------------- # Stop loss # ----------------------------------------------------------------------- def _determine_stop_loss(self, b_sl, sig_sl, support, price): """Determine the best stop loss price from available data.""" candidates = [] if b_sl > 0: candidates.append(b_sl) if sig_sl > 0: candidates.append(sig_sl) if support > 0: # SL slightly below support candidates.append(support * 0.97) if not candidates: # Fallback: 8% below current price return round(price * 0.92, 2) # Use the tightest reasonable SL best = max(candidates) # Ensure SL is below current price if best >= price: best = price * 0.95 return round(best, 2) # ----------------------------------------------------------------------- # Take profit ladder # ----------------------------------------------------------------------- def _calc_tp_ladder( self, tp1, tp2, resistance, sig_target, ml_target, price, ath, ): """Build a TP ladder with TP1, TP2, and resistance targets.""" ladder = [] # TP1: conservative target tp1_price = tp1 if tp1 > price else 0 if tp1_price <= 0 and sig_target > price: tp1_price = price + (sig_target - price) * 0.5 if tp1_price <= 0: tp1_price = price * 1.05 # TP2: moderate target tp2_price = tp2 if tp2 > price else 0 if tp2_price <= 0 and sig_target > price: tp2_price = sig_target if tp2_price <= 0 and ml_target > price: tp2_price = ml_target if tp2_price <= 0: tp2_price = price * 1.12 # TP3: resistance / aggressive target tp3_price = 0 if resistance > price: tp3_price = resistance elif ath > price: tp3_price = price + (ath - price) * 0.5 if tp3_price <= 0: tp3_price = price * 1.20 # Ensure TP1 < TP2 < TP3 targets = sorted([tp1_price, tp2_price, tp3_price]) labels = ['TP1_CONSERVATIVE', 'TP2_MODERATE', 'TP3_AGGRESSIVE'] sell_pcts = [40.0, 35.0, 25.0] for i, (tp, label, sell_pct) in enumerate(zip(targets, labels, sell_pcts)): gain_pct = ((tp - price) / price * 100) if price > 0 else 0 ladder.append({ 'level': i + 1, 'label': label, 'price': round(tp, 2), 'gain_pct': round(gain_pct, 2), 'sell_pct': sell_pct, }) return ladder # ----------------------------------------------------------------------- # Risk:Reward calculation # ----------------------------------------------------------------------- def _calc_risk_reward(self, avg_entry, stop_loss, tp_ladder): """Calculate risk:reward ratio for each TP level and overall.""" if avg_entry <= 0 or stop_loss <= 0 or stop_loss >= avg_entry: return { 'ratio': 0.0, 'risk_amount': 0.0, 'levels': [], } risk = avg_entry - stop_loss levels = [] total_weighted_rr = 0.0 total_weight = 0.0 for tp in tp_ladder: tp_price = tp.get('price', 0) sell_pct = tp.get('sell_pct', 0) if tp_price > avg_entry and risk > 0: reward = tp_price - avg_entry rr = round(reward / risk, 2) levels.append({ 'label': tp.get('label', ''), 'price': tp_price, 'reward': round(reward, 2), 'rr_ratio': rr, }) total_weighted_rr += rr * (sell_pct / 100.0) total_weight += sell_pct / 100.0 overall_rr = round( total_weighted_rr / total_weight if total_weight > 0 else 0, 2, ) return { 'ratio': overall_rr, 'risk_amount': round(risk, 2), 'risk_pct': round((risk / avg_entry) * 100, 2) if avg_entry > 0 else 0, 'levels': levels, } # ----------------------------------------------------------------------- # Position sizing # ----------------------------------------------------------------------- def _calc_position_size(self, avg_entry, stop_loss, sig_conf, bull_score): """Suggest position size based on risk management rules.""" if avg_entry <= 0 or stop_loss <= 0 or stop_loss >= avg_entry: return self._default_position_sizing() risk_per_unit = avg_entry - stop_loss risk_pct_per_unit = (risk_per_unit / avg_entry) * 100 # Adjust max risk based on confidence confidence_factor = max(0.3, min(1.0, (sig_conf + bull_score / 100.0) / 2.0)) adjusted_risk_pct = self.MAX_RISK_PCT * confidence_factor # Position size in IDR max_risk_idr = self.DEFAULT_CAPITAL * (adjusted_risk_pct / 100.0) if risk_per_unit > 0: max_units = max_risk_idr / risk_per_unit position_value = max_units * avg_entry else: position_value = self.DEFAULT_CAPITAL * 0.05 # Cap at reasonable percentage of portfolio position_value = min(position_value, self.DEFAULT_CAPITAL * 0.15) # Allocation per scaling level level_allocations = [] for i, ratio in enumerate(self.SCALING_RATIOS): level_allocations.append({ 'level': i + 1, 'allocation_idr': round(position_value * ratio, 0), 'allocation_pct': round(ratio * 100, 1), }) # Risk tier if risk_pct_per_unit <= 3: risk_tier = 'CONSERVATIVE' elif risk_pct_per_unit <= 6: risk_tier = 'MODERATE' elif risk_pct_per_unit <= 10: risk_tier = 'AGGRESSIVE' else: risk_tier = 'HIGH_RISK' return { 'suggested_total_idr': round(position_value, 0), 'portfolio_allocation_pct': round( (position_value / self.DEFAULT_CAPITAL) * 100, 2, ), 'max_risk_idr': round(max_risk_idr, 0), 'max_risk_pct': round(adjusted_risk_pct, 2), 'risk_per_unit_pct': round(risk_pct_per_unit, 2), 'confidence_factor': round(confidence_factor, 2), 'risk_tier': risk_tier, 'level_allocations': level_allocations, 'reference_capital_idr': self.DEFAULT_CAPITAL, } def _default_position_sizing(self): return { 'suggested_total_idr': 0, 'portfolio_allocation_pct': 0.0, 'max_risk_idr': 0, 'max_risk_pct': 0.0, 'risk_per_unit_pct': 0.0, 'confidence_factor': 0.0, 'risk_tier': 'UNKNOWN', 'level_allocations': [], 'reference_capital_idr': self.DEFAULT_CAPITAL, } # ----------------------------------------------------------------------- # Opportunity score # ----------------------------------------------------------------------- def _calc_opportunity_score( self, rr, sig_conf, dir_prob, bull_score, phase, mom_state, mtf, upside, range_pos, zscore, mr_conf, ): """Calculate overall opportunity score (0-100).""" score = 0.0 # Risk:Reward quality – 20 pts rr_ratio = rr.get('ratio', 0) if isinstance(rr, dict) else rr if rr_ratio >= 4.0: score += 20.0 elif rr_ratio >= 3.0: score += 17.0 elif rr_ratio >= 2.5: score += 14.0 elif rr_ratio >= 2.0: score += 11.0 elif rr_ratio >= 1.5: score += 8.0 elif rr_ratio >= 1.0: score += 5.0 elif rr_ratio > 0: score += 2.0 # Signal confidence – 15 pts score += _clamp(sig_conf * 15.0, 0, 15) # Direction probability – 10 pts score += _clamp(dir_prob * 10.0, 0, 10) # Bullish score – 10 pts score += _clamp((bull_score / 100.0) * 10.0, 0, 10) # Phase quality – 10 pts phase_scores = { 'EARLY_BULLISH': 10.0, 'ACCUMULATION': 8.0, 'PEAK_BULLISH': 6.0, 'NOT_BULLISH': 3.0, 'MARKDOWN': 1.0, } score += phase_scores.get(phase, 3.0) # Momentum state – 8 pts mom_scores = { 'ACCELERATING_UP': 8.0, 'REVERSING_UP': 7.0, 'DECELERATING_UP': 5.0, 'STABLE': 4.0, 'DECELERATING_DOWN': 2.0, 'ACCELERATING_DOWN': 1.0, 'REVERSING_DOWN': 1.5, } score += mom_scores.get(mom_state, 3.0) # MTF confirmation – 7 pts if mtf: score += 7.0 # Upside potential – 8 pts if upside > 30: score += 8.0 elif upside > 20: score += 6.0 elif upside > 10: score += 4.0 elif upside > 5: score += 2.0 # Entry quality from range position – 7 pts _rp = range_pos if range_pos is not None else 50.0 if 0 < _rp <= 25: score += 7.0 elif _rp <= 40: score += 5.0 elif _rp <= 55: score += 3.0 else: score += 1.0 # MR confidence bonus – 5 pts score += _clamp(mr_conf * 5.0, 0, 5) return round(_clamp(score, 0, 100), 1) # ----------------------------------------------------------------------- # Position type recommendation # ----------------------------------------------------------------------- def _recommend_position_type( self, phase, range_pos, zscore, sig_type, range_width, is_mr, ): """Recommend the position type based on conditions.""" _rp = range_pos if range_pos is not None else 50.0 if phase == 'ACCUMULATION' and zscore < -0.5: return 'ACCUMULATION_DCA' if phase == 'EARLY_BULLISH' and _rp < 40: return 'TREND_FOLLOW' if is_mr and range_width >= 15 and _rp < 30: return 'RANGE_TRADE_LONG' if sig_type and sig_type.lower() in ('buy', 'strong_buy'): return 'SIGNAL_ENTRY' if phase in ('EARLY_BULLISH', 'PEAK_BULLISH'): return 'MOMENTUM_RIDE' return 'CAUTIOUS_ENTRY' # ----------------------------------------------------------------------- # Signal filter # ----------------------------------------------------------------------- def _passes_signal_filter(self, item, signal_status): """Check if item passes the signal status filter.""" if signal_status == 'all': return True if signal_status == 'active_buy': sig_status = item.get('signal_status', '') sig_type = item.get('signal_type', '') return ( sig_status and sig_status.lower() == 'active' and sig_type and sig_type.lower() in ('buy', 'strong_buy', 'accumulate') ) if signal_status == 'high_confidence': return item.get('signal_confidence', 0) >= 0.7 return True # ----------------------------------------------------------------------- # Stats # ----------------------------------------------------------------------- def _compute_stats(self, scored): total = len(scored) if total == 0: return self._empty_stats() scores = [s['opportunity_score'] for s in scored] rr_ratios = [s['risk_reward_ratio'] for s in scored] avg_score = round(sum(scores) / total, 1) avg_rr = round(sum(rr_ratios) / total, 2) if rr_ratios else 0.0 grade_counts = {} for s in scored: g = s.get('grade', 'F') grade_counts[g] = grade_counts.get(g, 0) + 1 type_counts = {} for s in scored: pt = s.get('position_type', 'UNKNOWN') type_counts[pt] = type_counts.get(pt, 0) + 1 high_opp = sum(1 for s in scored if s['opportunity_score'] >= 70) high_opp_pct = round((high_opp / total) * 100, 1) good_rr = sum(1 for s in scored if s['risk_reward_ratio'] >= 2.0) good_rr_pct = round((good_rr / total) * 100, 1) top_opportunities = sorted(scored, key=lambda x: x['opportunity_score'], reverse=True)[:5] top_opp_summary = [ { 'symbol': t['symbol'], 'opportunity_score': t['opportunity_score'], 'grade': t['grade'], 'risk_reward_ratio': t['risk_reward_ratio'], 'position_type': t['position_type'], } for t in top_opportunities ] return { 'total': total, 'avg_opportunity_score': avg_score, 'avg_risk_reward_ratio': avg_rr, 'top_score': max(scores), 'bottom_score': min(scores), 'high_opportunity_count': high_opp, 'high_opportunity_pct': high_opp_pct, 'good_rr_count': good_rr, 'good_rr_pct': good_rr_pct, 'grade_distribution': grade_counts, 'position_type_distribution': type_counts, 'top_opportunities': top_opp_summary, } # ----------------------------------------------------------------------- # Sorting # ----------------------------------------------------------------------- def _sort_items(self, items, sort_by): key_name, desc = self.SORT_OPTIONS.get(sort_by, ('opportunity_score', True)) try: items.sort( key=lambda x: _safe_float(x.get(key_name, 0)), reverse=desc, ) except Exception: pass return items # ----------------------------------------------------------------------- # Helpers # ----------------------------------------------------------------------- @staticmethod def _empty_result(page=1, limit=50): return { 'items': [], 'total': 0, 'page': page, 'limit': limit, 'has_more': False, 'stats': PositionArchitectService._empty_stats(), } @staticmethod def _empty_stats(): return { 'total': 0, 'avg_opportunity_score': 0.0, 'avg_risk_reward_ratio': 0.0, 'top_score': 0.0, 'bottom_score': 0.0, 'high_opportunity_count': 0, 'high_opportunity_pct': 0.0, 'good_rr_count': 0, 'good_rr_pct': 0.0, 'grade_distribution': {}, 'position_type_distribution': {}, 'top_opportunities': [], }