""" V2 交易计划生成器 - 5维度综合评分 + 多周期共振分析 """ import json import logging from collections import Counter from pathlib import Path from typing import Dict, List, Optional, Tuple from datetime import datetime from sqlalchemy.orm import Session from app.services.cache import get_cached_data from app.analysis_models import ( ReviewPlanV2, SymbolScoreV2, TradingPlanV2, SectorHeat, ) logger = logging.getLogger(__name__) # 品种配置加载 CONFIG_DIR = Path(__file__).resolve().parent.parent.parent / "config" # 板块分类 SECTOR_MAP = { "贵金属": ["沪银", "沪金"], "有色金属": ["沪铜", "沪锌", "沪锡", "沪铅", "沪铝", "铝合金", "沪镍"], "黑色系": ["焦炭", "焦煤", "螺纹钢", "热卷", "铁矿石"], "能源": ["原油", "低硫燃油", "燃油"], "化工": ["PTA", "甲醇", "尿素", "PVC", "乙二醇", "合成橡胶", "烧碱", "纯碱", "玻璃", "橡胶"], "农产品": ["白糖", "棉花", "棕榈油", "豆粕"], "新能源": ["氧化铝", "多晶硅", "碳酸锂", "工业硅"], "股指": ["中证1000"], "航运": ["集运欧线"], "橡胶": ["20号胶"], } # 权重 WEIGHTS = { "amplitude": 0.20, "volume": 0.15, "change": 0.15, "trend": 0.35, "activity": 0.15, } def load_symbols_config() -> Dict[str, str]: """加载品种配置 {中文名: 合约代码}""" config_path = CONFIG_DIR / "symbols_config.json" if not config_path.exists(): logger.warning(f"品种配置文件不存在: {config_path}") return {} with open(config_path, "r", encoding="utf-8") as f: data = json.load(f) return data.get("futures", {}) def _get_candle_date(candle: dict) -> Optional[str]: """从K线数据中提取日期字符串 (YYYY-MM-DD)""" dt_str = candle.get("datetime") or candle.get("time") if not dt_str: return None try: if isinstance(dt_str, str): dt = datetime.fromisoformat(dt_str.replace("Z", "+00:00")) else: dt = dt_str return dt.strftime("%Y-%m-%d") except Exception: return None def _get_candle_values(candles: List[dict]) -> Tuple[list, list, list, list, list]: """从K线列表提取 OHLCV""" opens, highs, lows, closes, volumes = [], [], [], [], [] for c in candles: opens.append(float(c.get("open", 0))) highs.append(float(c.get("high", 0))) lows.append(float(c.get("low", 0))) closes.append(float(c.get("close", 0))) volumes.append(float(c.get("volume", 0))) return opens, highs, lows, closes, volumes def _calc_single_period_trend(candles: List[dict]) -> float: """ 计算单周期趋势分 (-50 ~ +50) MA排列: 多头+50 / 空头-50 MACD: DIF>DEA +30 / DIF ma20: score += 50 else: score -= 50 # MACD if macd_dif > macd_dea: score += 30 else: score -= 30 # 归一化到 -50 ~ +50 return max(-50, min(50, score * 50 / 80)) def _calc_trend_score(t60: float, t15: float, t5: float) -> Tuple[float, str, str]: """ 三周期共振判定 Returns: (趋势总分 0-100, 方向标签, 方向简写) """ # 共振判定 if t60 > 20 and t15 > 20 and t5 > 20: direction = "多头共振" tag = "强多" raw = 80 + min(20, (t60 + t15 + t5) / 3 * 20 / 50) elif t60 < -20 and t15 < -20 and t5 < -20: direction = "空头共振" tag = "强空" raw = 80 + min(20, abs(t60 + t15 + t5) / 3 * 20 / 50) elif t60 > 0 and t15 > 0: direction = "偏多震荡" tag = "偏多" raw = 50 + min(20, (t60 + t15) / 2 * 20 / 50) elif t60 < 0 and t15 < 0: direction = "偏空震荡" tag = "偏空" raw = 50 + min(20, abs(t60 + t15) / 2 * 20 / 50) else: direction = "多空交织" tag = "震荡" raw = max(0, 40 - abs(t60 + t15 + t5) / 3) return round(raw, 1), direction, tag def _get_price_precision(price: float) -> int: """根据价格大小确定合适的小数位数""" if price >= 10000: return 0 # 如白银8250、股指5000+ elif price >= 1000: return 1 # 如原油528、黄金685(实际2位) elif price >= 100: return 2 # 如大部分化工品 else: return 2 # 小数值的品种 def _round_by_price(value: float, price: float) -> float: """根据价格精度四舍五入""" precision = _get_price_precision(price) return round(value, precision) def _calc_pivot_points(high: float, low: float, close: float) -> dict: """计算枢轴点和支撑/阻力位（根据价格精度自动调整小数位）""" pivot = (high + low + close) / 3 r1 = 2 * pivot - low s1 = 2 * pivot - high r2 = pivot + (high - low) s2 = pivot - (high - low) # 根据价格精度四舍五入 p = _get_price_precision(close) return { "pivot": round(pivot, p), "r1": round(r1, p), "r2": round(r2, p), "s1": round(s1, p), "s2": round(s2, p), } def _normalize_scores(values: List[float], reverse: bool = False) -> List[float]: """将一组值归一化到 0-100""" if not values: return [] min_v = min(values) max_v = max(values) if max_v == min_v: return [50.0] * len(values) result = [] for v in values: if reverse: score = (max_v - v) / (max_v - min_v) * 100 else: score = (v - min_v) / (max_v - min_v) * 100 result.append(round(score, 1)) return result def _normalize_abs_scores(values: List[float]) -> List[float]: """将绝对值归一化到 0-100（涨跌幅用绝对值排名）""" abs_values = [abs(v) for v in values] return _normalize_scores(abs_values) def _generate_trigger(score_data: dict, direction: str) -> str: """ 根据分析结果动态生成触发条件 Args: score_data: 品种评分数据 direction: 方向 (long/short) Returns: 触发条件文本 """ conditions = [] direction_tag = score_data.get("direction_tag", "震荡") volume_ratio = score_data.get("volume_ratio", 1.0) amplitude_pct = score_data.get("amplitude_pct", 0) change_pct = score_data.get("change_pct", 0) # 根据方向标签生成基础条件 if direction == "long": if direction_tag == "强多": conditions.append("价格站稳支撑位上方") elif direction_tag == "偏多": conditions.append("回踩支撑企稳") else: conditions.append("突破枢轴点P位") else: if direction_tag == "强空": conditions.append("价格跌破阻力位下方") elif direction_tag == "偏空": conditions.append("反弹阻力受阻") else: conditions.append("跌破枢轴点P位") # 根据量比添加量能条件 if volume_ratio > 1.5: conditions.append("量能持续放大") elif volume_ratio > 1.2: conditions.append("5分钟K线放量确认") else: conditions.append("等待量能配合") # 根据振幅添加波动条件 if amplitude_pct > 2: conditions.append("注意波动风险") elif amplitude_pct > 1: conditions.append("关注区间突破") # 根据涨跌幅添加趋势条件 if abs(change_pct) > 2: if change_pct > 0: conditions.append("强势延续确认") else: conditions.append("弱势延续确认") return " + ".join(conditions[:3]) # 最多返回3个条件 def generate_plan(db: Session, review_date_str: str, week_day: str) -> dict: """ 生成V2复盘与交易计划 Returns: { "review_date_id": int, "summary": {...}, "scores": [...], "plans": [...], "sectors": [...], } """ symbols_config = load_symbols_config() if not symbols_config: raise ValueError("品种配置文件为空") # 反转映射: 合约代码 -> 中文名 code_to_name = {v: k for k, v in symbols_config.items()} all_symbols = list(symbols_config.values()) logger.info(f"开始生成交易计划: {review_date_str} ({week_day}), 共 {len(all_symbols)} 个品种") # 将复盘日期作为数据截止时间（当天 23:59:59） try: end_time = datetime.strptime(review_date_str, "%Y-%m-%d").replace(hour=23, minute=59, second=59) except ValueError: end_time = datetime.now() # ==================== 第一步: 数据收集 ==================== symbol_data = {} periods_needed = ["daily", "60min", "15min", "5min"] for symbol in all_symbols: try: data = get_cached_data( db, symbol, "futures", periods=periods_needed, end_time=end_time, max_candles=100, ) if data and data.get("timeframes"): symbol_data[symbol] = data else: logger.warning(f"{symbol} 无数据") except Exception as e: logger.error(f"获取 {symbol} 数据失败: {e}") if not symbol_data: raise ValueError("无法从数据库获取任何品种数据，请先进行数据更新") logger.info(f"成功获取 {len(symbol_data)}/{len(all_symbols)} 个品种数据") # ==================== 第二步: 多维度打分 ==================== raw_scores = [] for symbol, data in symbol_data.items(): name = code_to_name.get(symbol, symbol) tf = data.get("timeframes", {}) daily = tf.get("daily", []) h1 = tf.get("60min", []) m15 = tf.get("15min", []) m5 = tf.get("5min", []) # 日线数据 if not daily or len(daily) < 2: continue _, d_highs, d_lows, d_closes, d_volumes = _get_candle_values(daily) today_high = d_highs[-1] today_low = d_lows[-1] today_close = d_closes[-1] prev_close = d_closes[-2] if len(d_closes) >= 2 else today_close today_volume = d_volumes[-1] # 提取实际数据日期（日线最后一根K线的日期） data_date = _get_candle_date(daily[-1]) # 近5日均量 recent_vols = d_volumes[-6:-1] if len(d_volumes) >= 6 else d_volumes[:-1] avg_vol_5 = sum(recent_vols) / len(recent_vols) if recent_vols else today_volume # A. 振幅 amplitude_pct = (today_high - today_low) / today_close * 100 if today_close else 0 # B. 量比 volume_ratio = today_volume / avg_vol_5 if avg_vol_5 > 0 else 1.0 # C. 涨跌幅 change_pct = (today_close - prev_close) / prev_close * 100 if prev_close else 0 # D. 多周期趋势 t60 = _calc_single_period_trend(h1) if h1 else 0 t15 = _calc_single_period_trend(m15) if m15 else 0 t5 = _calc_single_period_trend(m5) if m5 else 0 trend_total, direction, direction_tag = _calc_trend_score(t60, t15, t5) # E. 活跃度 (5min K线近10根) if m5 and len(m5) >= 5: recent_m5 = m5[-10:] if len(m5) >= 10 else m5 _, _, _, m5_closes, m5_vols = _get_candle_values(recent_m5) # 价格变动率 if m5_closes and m5_closes[0] > 0: price_change_rate = abs(m5_closes[-1] - m5_closes[0]) / m5_closes[0] * 100 else: price_change_rate = 0 # 量能活跃度 avg_m5_vol = sum(m5_vols) / len(m5_vols) if m5_vols else 0 vol_activity = min(100, avg_m5_vol / 100) if avg_m5_vol > 0 else 0 price_activity = min(100, price_change_rate * 50) activity_score = (vol_activity + price_activity) / 2 else: activity_score = 0 # 关键点位 pivots = _calc_pivot_points(today_high, today_low, today_close) raw_scores.append({ "symbol": symbol, "name": name, "data_date": data_date, "close_price": today_close, "prev_close": prev_close, "high_price": today_high, "low_price": today_low, "volume": today_volume, "avg_volume_5": avg_vol_5, "amplitude_pct": round(amplitude_pct, 2), "volume_ratio": round(volume_ratio, 2), "change_pct": round(change_pct, 2), "amplitude_raw": amplitude_pct, "volume_raw": volume_ratio, "change_raw": abs(change_pct), "trend_score": trend_total, "trend_60m": round(t60, 1), "trend_15m": round(t15, 1), "trend_5m": round(t5, 1), "direction": direction, "direction_tag": direction_tag, "activity_raw": activity_score, "pivots": pivots, }) if not raw_scores: raise ValueError("没有足够的数据进行评分计算") # ==================== 第三步: 归一化评分 ==================== amp_scores = _normalize_scores([s["amplitude_raw"] for s in raw_scores]) vol_scores = _normalize_scores([s["volume_raw"] for s in raw_scores]) chg_scores = _normalize_abs_scores([s["change_raw"] for s in raw_scores]) act_scores = _normalize_scores([s["activity_raw"] for s in raw_scores]) for i, s in enumerate(raw_scores): s["amplitude_score"] = amp_scores[i] s["volume_score"] = vol_scores[i] s["change_score"] = chg_scores[i] s["activity_score"] = act_scores[i] # 综合评分 = 振幅×0.20 + 量能×0.15 + 涨跌幅×0.15 + 趋势×0.35 + 活跃度×0.15 # 趋势分映射: trend_score 0-100 直接使用 s["composite_score"] = round( amp_scores[i] * WEIGHTS["amplitude"] + vol_scores[i] * WEIGHTS["volume"] + chg_scores[i] * WEIGHTS["change"] + s["trend_score"] * WEIGHTS["trend"] + act_scores[i] * WEIGHTS["activity"], 1 ) # 排序 raw_scores.sort(key=lambda x: x["composite_score"], reverse=True) # ==================== 第四步: 分类 ==================== for rank, s in enumerate(raw_scores, 1): s["rank"] = rank score = s["composite_score"] trend = s["trend_score"] if score >= 55 and trend >= 50: s["category"] = "green" elif score >= 45 or (score >= 40 and 30 <= trend < 50): s["category"] = "yellow" else: s["category"] = "red" # ==================== 第五步: 生成交易计划 ==================== plans = [] green_items = [s for s in raw_scores if s["category"] == "green"] for s in green_items: # 根据方向标签判断实际方向（强多/偏多→做多，强空/偏空→做空） direction_tag = s.get("direction_tag", "震荡") if direction_tag in ["强空", "偏空"]: direction = "short" elif direction_tag in ["强多", "偏多"]: direction = "long" else: # 震荡情况下根据涨跌幅判断 direction = "long" if s["change_pct"] > 0 else "short" pivots = s["pivots"] price = s["close_price"] if direction == "long": entry_low = _round_by_price(pivots["s1"], price) entry_high = _round_by_price(pivots["pivot"], price) stop_loss = _round_by_price(pivots["s2"], price) target1 = _round_by_price(pivots["r1"], price) target2 = _round_by_price(pivots["r2"], price) else: entry_low = _round_by_price(pivots["pivot"], price) entry_high = _round_by_price(pivots["r1"], price) stop_loss = _round_by_price(pivots["r2"], price) target1 = _round_by_price(pivots["s1"], price) target2 = _round_by_price(pivots["s2"], price) # 根据分析结果动态生成触发条件 trigger = _generate_trigger(s, direction) plans.append({ "symbol": s["symbol"], "name": s["name"], "direction": direction, "composite_score": s["composite_score"], "entry_low": entry_low, "entry_high": entry_high, "stop_loss": stop_loss, "target1": target1, "target2": target2, "trigger": trigger, "amplitude_score": s["amplitude_score"], "volume_score": s["volume_score"], "trend_score": s["trend_score"], "activity_score": s["activity_score"], "category": s["category"], }) # ==================== 第六步: 板块热度 ==================== sectors = _calc_sector_heat(raw_scores, code_to_name) # ==================== 第七步: 汇总 ==================== bull_count = sum(1 for s in raw_scores if s["trend_score"] >= 50) bear_count = sum(1 for s in raw_scores if s["trend_score"] < 30) neutral_count = len(raw_scores) - bull_count - bear_count # 统计实际数据日期（取出现最多的日期作为基准数据日期） date_counts = Counter(s.get("data_date") for s in raw_scores if s.get("data_date")) actual_data_date = date_counts.most_common(1)[0][0] if date_counts else review_date_str # 判断数据日期是否与复盘日期一致 data_date_matches = (actual_data_date == review_date_str) # 核心结论 top3 = raw_scores[:3] top_names = [f"{s['symbol']} {s['name']}" for s in top3] main_direction = "多头" if bull_count > bear_count else "空头" if bear_count > bull_count else "震荡" core_conclusion = f"{main_direction}格局{'延续' if bull_count > 15 else '分化'}，{'贵金属+黑色系共振做多' if bull_count > 20 else '板块轮动明显'}" # 风险提示 risk_warnings = [] if top3: top_sym = top3[0] risk_warnings.append( f"极值品种严禁追{'多' if top_sym['trend_score'] >= 50 else '空'}：" f"{top_sym['symbol']}({top_sym['composite_score']}分) " f"{'周一大概率高开' if top_sym['change_pct'] > 2 else '注意风险控制'}" ) if bear_count > 0: bear_syms = [s for s in raw_scores if s["category"] == "red" and s["trend_score"] < 30][:2] if bear_syms: names = "/".join(s["symbol"] for s in bear_syms) risk_warnings.append(f"空头品种谨慎：{names} 趋势偏空，需严格止损") risk_warnings.append("关注宏观数据和消息面变化，可能影响开盘方向") if any(s["volume_ratio"] > 2.5 for s in raw_scores): high_vol = [s for s in raw_scores if s["volume_ratio"] > 2.5] risk_warnings.append( f"量能异常品种：{', '.join(s['symbol'] for s in high_vol[:3])} " f"量比超2.5倍，需确认量能可持续性" ) # 构建数据基准描述 if data_date_matches: data_basis = f"{actual_data_date} 收盘 ({len(symbol_data)} 品种) | 多周期共振分析" else: data_basis = f"复盘日期: {review_date_str} | 数据截至: {actual_data_date} 收盘 ({len(symbol_data)} 品种) | 多周期共振分析" return { "review_date": review_date_str, "week_day": week_day, "actual_data_date": actual_data_date, "data_date_matches": data_date_matches, "data_basis": data_basis, "core_conclusion": core_conclusion, "bull_count": bull_count, "bear_count": bear_count, "neutral_count": neutral_count, "opportunity_count": len(green_items), "risk_warnings": risk_warnings, "scores": raw_scores, "plans": plans, "sectors": sectors, "symbol_count": len(symbol_data), } def _calc_sector_heat(scores: List[dict], code_to_name: dict) -> List[dict]: """计算板块热度""" # 构建 symbol -> score 映射 name_to_score = {} for s in scores: name_to_score[s["name"]] = s sectors = [] for sector_name, members_names in SECTOR_MAP.items(): members = [] for mn in members_names: if mn in name_to_score: sc = name_to_score[mn] members.append({ "symbol": sc["symbol"], "name": sc["name"], "score": sc["composite_score"], "trend": sc["trend_score"], "change_pct": sc["change_pct"], }) if not members: continue avg_score = round(sum(m["score"] for m in members) / len(members), 1) avg_trend = round(sum(m["trend"] for m in members) / len(members), 0) if avg_trend > 20: direction = "多头" elif avg_trend < -20: direction = "空头" else: direction = "震荡" # 热度等级 if avg_score >= 75: heat = 3 elif avg_score >= 60: heat = 2 elif avg_score >= 45: heat = 1 else: heat = 0 # 龙头 leader = max(members, key=lambda m: m["score"]) sectors.append({ "sector_name": sector_name, "avg_score": avg_score, "avg_trend": int(avg_trend), "direction": direction, "heat_level": heat, "leader_symbol": leader["symbol"], "leader_score": leader["score"], "members": sorted(members, key=lambda m: m["score"], reverse=True), }) sectors.sort(key=lambda s: s["avg_score"], reverse=True) return sectors def save_plan_to_db(db: Session, plan_data: dict) -> int: """将计划数据保存到数据库""" review_date_str = plan_data["review_date"] week_day = plan_data["week_day"] # 清理旧数据 existing = db.query(ReviewPlanV2).filter_by(review_date=review_date_str).first() if existing: review_plan_id = existing.id # 清理关联数据 db.query(SymbolScoreV2).filter_by(review_date_id=review_plan_id).delete() db.query(TradingPlanV2).filter_by(review_date_id=review_plan_id).delete() db.query(SectorHeat).filter_by(review_date_id=review_plan_id).delete() # 更新元数据 existing.week_day = week_day existing.data_basis = plan_data.get("data_basis") existing.core_conclusion = plan_data.get("core_conclusion") existing.bull_count = plan_data.get("bull_count") existing.bear_count = plan_data.get("bear_count") existing.neutral_count = plan_data.get("neutral_count") existing.opportunity_count = plan_data.get("opportunity_count") existing.risk_warnings = plan_data.get("risk_warnings") existing.actual_data_date = plan_data.get("actual_data_date") existing.data_date_matches = 1 if plan_data.get("data_date_matches") else 0 else: review_plan = ReviewPlanV2( review_date=review_date_str, week_day=week_day, data_basis=plan_data.get("data_basis"), core_conclusion=plan_data.get("core_conclusion"), bull_count=plan_data.get("bull_count"), bear_count=plan_data.get("bear_count"), neutral_count=plan_data.get("neutral_count"), opportunity_count=plan_data.get("opportunity_count"), risk_warnings=plan_data.get("risk_warnings"), actual_data_date=plan_data.get("actual_data_date"), data_date_matches=1 if plan_data.get("data_date_matches") else 0, ) db.add(review_plan) db.flush() review_plan_id = review_plan.id # 保存评分数据 for s in plan_data.get("scores", []): pivots = s.get("pivots", {}) score_record = SymbolScoreV2( review_date_id=review_plan_id, symbol=s["symbol"], name=s["name"], close_price=s.get("close_price"), prev_close=s.get("prev_close"), high_price=s.get("high_price"), low_price=s.get("low_price"), volume=s.get("volume"), avg_volume_5=s.get("avg_volume_5"), amplitude_score=s.get("amplitude_score"), volume_score=s.get("volume_score"), change_score=s.get("change_score"), trend_score=s.get("trend_score"), activity_score=s.get("activity_score"), composite_score=s.get("composite_score"), amplitude_pct=s.get("amplitude_pct"), change_pct=s.get("change_pct"), volume_ratio=s.get("volume_ratio"), trend_60m=s.get("trend_60m"), trend_15m=s.get("trend_15m"), trend_5m=s.get("trend_5m"), direction=s.get("direction"), direction_tag=s.get("direction_tag"), category=s.get("category"), pivot=pivots.get("pivot"), r1=pivots.get("r1"), r2=pivots.get("r2"), s1=pivots.get("s1"), s2=pivots.get("s2"), rank=s.get("rank"), data_date=s.get("data_date"), ) db.add(score_record) # 保存交易计划 for p in plan_data.get("plans", []): plan_record = TradingPlanV2( review_date_id=review_plan_id, symbol=p["symbol"], name=p["name"], direction=p["direction"], composite_score=p.get("composite_score"), entry_low=p.get("entry_low"), entry_high=p.get("entry_high"), stop_loss=p.get("stop_loss"), target1=p.get("target1"), target2=p.get("target2"), trigger=p.get("trigger"), amplitude_score=p.get("amplitude_score"), volume_score=p.get("volume_score"), trend_score=p.get("trend_score"), activity_score=p.get("activity_score"), category=p.get("category"), ) db.add(plan_record) # 保存板块热度 for sec in plan_data.get("sectors", []): sector_record = SectorHeat( review_date_id=review_plan_id, sector_name=sec["sector_name"], avg_score=sec.get("avg_score"), avg_trend=sec.get("avg_trend"), direction=sec.get("direction"), heat_level=sec.get("heat_level"), leader_symbol=sec.get("leader_symbol"), leader_score=sec.get("leader_score"), members=sec.get("members"), ) db.add(sector_record) db.commit() logger.info(f"交易计划已保存: {review_date_str}, ID={review_plan_id}") return review_plan_id def get_plan_data(db: Session, review_plan_id: int) -> Optional[dict]: """从数据库读取完整的计划数据""" plan_meta = db.query(ReviewPlanV2).filter_by(id=review_plan_id).first() if not plan_meta: return None scores = db.query(SymbolScoreV2).filter_by( review_date_id=review_plan_id ).order_by(SymbolScoreV2.composite_score.desc()).all() plans = db.query(TradingPlanV2).filter_by( review_date_id=review_plan_id ).order_by(TradingPlanV2.composite_score.desc()).all() sectors = db.query(SectorHeat).filter_by( review_date_id=review_plan_id ).order_by(SectorHeat.avg_score.desc()).all() return { "meta": { "id": plan_meta.id, "review_date": plan_meta.review_date, "week_day": plan_meta.week_day, "data_basis": plan_meta.data_basis, "actual_data_date": plan_meta.actual_data_date, "data_date_matches": bool(plan_meta.data_date_matches), "core_conclusion": plan_meta.core_conclusion, "bull_count": plan_meta.bull_count, "bear_count": plan_meta.bear_count, "neutral_count": plan_meta.neutral_count, "opportunity_count": plan_meta.opportunity_count, "risk_warnings": plan_meta.risk_warnings, "created_at": plan_meta.created_at.isoformat() if plan_meta.created_at else None, }, "scores": [_score_to_dict(s) for s in scores], "plans": [_plan_to_dict(p) for p in plans], "sectors": [_sector_to_dict(s) for s in sectors], } def _score_to_dict(s: SymbolScoreV2) -> dict: return { "id": s.id, "symbol": s.symbol, "name": s.name, "data_date": s.data_date, "close_price": s.close_price, "prev_close": s.prev_close, "high_price": s.high_price, "low_price": s.low_price, "volume": s.volume, "avg_volume_5": s.avg_volume_5, "amplitude_score": s.amplitude_score, "volume_score": s.volume_score, "change_score": s.change_score, "trend_score": s.trend_score, "activity_score": s.activity_score, "composite_score": s.composite_score, "amplitude_pct": s.amplitude_pct, "change_pct": s.change_pct, "volume_ratio": s.volume_ratio, "trend_60m": s.trend_60m, "trend_15m": s.trend_15m, "trend_5m": s.trend_5m, "direction": s.direction, "direction_tag": s.direction_tag, "category": s.category, "pivot": s.pivot, "r1": s.r1, "r2": s.r2, "s1": s.s1, "s2": s.s2, "rank": s.rank, } def _plan_to_dict(p: TradingPlanV2) -> dict: return { "id": p.id, "symbol": p.symbol, "name": p.name, "direction": p.direction, "composite_score": p.composite_score, "entry_low": p.entry_low, "entry_high": p.entry_high, "stop_loss": p.stop_loss, "target1": p.target1, "target2": p.target2, "trigger": p.trigger, "amplitude_score": p.amplitude_score, "volume_score": p.volume_score, "trend_score": p.trend_score, "activity_score": p.activity_score, "category": p.category, } def _sector_to_dict(s: SectorHeat) -> dict: return { "id": s.id, "sector_name": s.sector_name, "avg_score": s.avg_score, "avg_trend": s.avg_trend, "direction": s.direction, "heat_level": s.heat_level, "leader_symbol": s.leader_symbol, "leader_score": s.leader_score, "members": s.members, }