"""动态策略选择器 根据市场温度和板块状态，纯规则选择当日策略 profile。 不再使用 LLM 或数据库配置覆盖。 """ import logging from pydantic import BaseModel from app.config import settings from app.data.models import MarketTemperature, SectorInfo logger = logging.getLogger(__name__) class StrategyProfile(BaseModel): strategy_id: str name: str description: str entry_signal_priority: list[str] score_weights: dict[str, float] min_score: float buy_threshold: float max_position_pct: float allow_trading: bool = True actionable_limit: int = 2 watch_limit: int = 4 target_focus_sectors: int = 2 market_stance: str = "" decision_note: str = "" notes: list[str] = [] feedback_applied: bool = False feedback_notes: list[str] = [] generated_by: str = "rules" def get_strategy_profile_by_id(strategy_id: str) -> StrategyProfile: normalized = strategy_id or "defensive_watch" if normalized == "trend_breakout": normalized = "breakout_attack" actionable_cap = max(0, settings.actionable_limit) watch_cap = max(0, settings.watch_limit) profiles = { "pre_market_ambush": StrategyProfile( strategy_id="pre_market_ambush", name="盘前埋伏", description="盘前选出缩量整理到位、回踩支撑、尚未启动的埋伏标的。", entry_signal_priority=["launch", "pullback", "breakout_confirm", "reversal", "breakout"], score_weights={"catalyst": 0.25, "theme_money": 0.22, "stock_money": 0.18, "emotion_role": 0.10, "timing": 0.25}, min_score=55, buy_threshold=58, max_position_pct=25, allow_trading=True, actionable_limit=min(3, actionable_cap), watch_limit=min(5, watch_cap), target_focus_sectors=3, market_stance="埋伏待发", decision_note="盘前选票重点看整理充分度和催化预期，不追已启动标的。", notes=["优先缩量整理到位+催化预期", "回踩支撑位附近的主线成分优先"], ), "breakout_attack": StrategyProfile( strategy_id="breakout_attack", name="主线突破", description="市场偏强，优先寻找主线板块内的突破和突破确认。", entry_signal_priority=["breakout", "breakout_confirm", "launch", "pullback", "reversal"], score_weights={"catalyst": 0.30, "theme_money": 0.25, "stock_money": 0.20, "emotion_role": 0.15, "timing": 0.10}, min_score=62, buy_threshold=66, max_position_pct=30, allow_trading=True, actionable_limit=min(3, actionable_cap), watch_limit=min(4, watch_cap), target_focus_sectors=2, market_stance="主线进攻", decision_note="只处理最强主线前排，不扩散到跟风和后排。", notes=["优先做主线早中期板块", "放量突破优先于回踩低吸"], ), "pullback_rotation": StrategyProfile( strategy_id="pullback_rotation", name="回踩轮动", description="市场震荡分化，优先做回踩支撑和板块轮动中的低吸确认。", entry_signal_priority=["pullback", "breakout_confirm", "launch", "breakout", "reversal"], score_weights={"catalyst": 0.25, "theme_money": 0.28, "stock_money": 0.20, "emotion_role": 0.12, "timing": 0.15}, min_score=60, buy_threshold=63, max_position_pct=20, allow_trading=True, actionable_limit=min(2, actionable_cap), watch_limit=min(5, watch_cap), target_focus_sectors=2, market_stance="轮动低吸", decision_note="先等回踩承接和板块回流，再决定是否出手。", notes=["降低追高仓位", "更看重位置安全和回踩承接"], ), "launch_probe": StrategyProfile( strategy_id="launch_probe", name="启动试错", description="市场偏弱，适合少量观察启动型和反转型机会，不做强追涨。", entry_signal_priority=["launch", "reversal", "pullback", "breakout_confirm", "breakout"], score_weights={"catalyst": 0.28, "theme_money": 0.22, "stock_money": 0.20, "emotion_role": 0.12, "timing": 0.18}, min_score=58, buy_threshold=61, max_position_pct=10, allow_trading=True, actionable_limit=min(1, actionable_cap), watch_limit=min(4, watch_cap), target_focus_sectors=1, market_stance="轻仓试错", decision_note="只有极少数启动确认标的值得小仓试错。", notes=["仅做小仓位试错", "突破型需要更强板块一致性才可介入"], ), "defensive_watch": StrategyProfile( strategy_id="defensive_watch", name="防守观察", description="市场退潮，系统以观察池为主，不主动扩大出手。", entry_signal_priority=["pullback", "launch", "reversal", "breakout_confirm", "breakout"], score_weights={"catalyst": 0.22, "theme_money": 0.25, "stock_money": 0.18, "emotion_role": 0.15, "timing": 0.20}, min_score=56, buy_threshold=64, max_position_pct=5, allow_trading=False, actionable_limit=0, watch_limit=min(3, watch_cap), target_focus_sectors=1, market_stance="防守观察", decision_note="今天不主动出手，只保留少量观察名单。", notes=["原则上只保留观察池", "等待市场温度修复后再转入主动进攻"], ), } return profiles.get(normalized, profiles["defensive_watch"]).model_copy(deep=True) async def select_strategy_profile( market_temp: MarketTemperature | None, hot_sectors: list[SectorInfo], intraday: bool, scan_session: str = "", ) -> StrategyProfile: """纯规则策略选择。盘前埋伏 session 走独立 profile。""" if scan_session == "pre_market_ambush" or scan_session == "pre_market": return get_strategy_profile_by_id("pre_market_ambush") return _select_rule_profile(market_temp, hot_sectors, intraday) def _select_rule_profile( market_temp: MarketTemperature | None, hot_sectors: list[SectorInfo], intraday: bool, ) -> StrategyProfile: temp = market_temp.temperature if market_temp else 0 early_count = sum(1 for s in hot_sectors[:5] if s.stage == "early") late_count = sum(1 for s in hot_sectors[:5] if s.stage in ("late", "end")) if temp >= 65 and early_count >= 1: return get_strategy_profile_by_id("breakout_attack") if temp >= 45 and late_count < 2: return get_strategy_profile_by_id("pullback_rotation") if temp >= 30: return get_strategy_profile_by_id("launch_probe") return get_strategy_profile_by_id("defensive_watch")