""" 结构化行情择时风控模块 专门处理A股结构化行情下的板块集中度风险和择时信号 功能： 1. 择时：单板块连续大涨提醒风险，连续大跌提示低吸 2. 风控：控制板块和风格集中度，不押注单一方向 3. 消息风险：热点板块灵敏度提高，提前预警利好出尽 Author: 关羽（云长） Date: 2026-03-27 """ from dataclasses import dataclass from typing import List, Dict, Tuple @dataclass class SectorInfo: """板块信息""" name: str style: str # 风格：AI/新能源/消费/周期等 recent_gain: float # 近期累计涨幅% position_ratio: float # 组合中该板块仓位比例 stock_count: int # 持有股票数量 is_hot: bool = False # 是否是当前热点板块 class StructuredMarketTiming: """结构化行情择时""" def __init__(self, high_position_threshold: float = 0.15, # 单板块仓位超15%提示减仓 sector_rally_warning: float = 30.0, # 累计涨30%提示风险 sector_rally_stop: float = 50.0, # 累计涨50%强制提高风控权重 sector_dump_opportunity: float = -20.0): # 跌20%提示机会 self.high_position_threshold = high_position_threshold self.sector_rally_warning = sector_rally_warning self.sector_rally_stop = sector_rally_stop self.sector_dump_opportunity = sector_dump_opportunity def timing_sector(self, sector: SectorInfo) -> Tuple[str, str, int]: """ 板块择时 返回：(信号类型, 建议, 风险增量评分) 信号类型: bull/bear/neutral """ risk_increment = 0 suggestion = "" signal = "neutral" # 单板块仓位过高提醒 if sector.position_ratio > self.high_position_threshold: risk_increment += int((sector.position_ratio - self.high_position_threshold) * 100) suggestion += f"\n⚠️ 单板块仓位{sector.position_ratio:.1%}，超过{self.high_position_threshold:.1%}阈值，建议适度减仓分散" signal = "bear" # 板块连续大涨提醒风险 if sector.recent_gain >= self.sector_rally_stop: risk_increment += 20 suggestion += f"\n🔴 板块累计涨幅{sector.recent_gain:.1f}%，超过{self.sector_rally_stop:.0f}%，警惕过热，建议整体止盈" signal = "bear" elif sector.recent_gain >= self.sector_rally_warning: risk_increment += 10 suggestion += f"\n🟡 板块累计涨幅{sector.recent_gain:.1f}%，已有较大涨幅，提高风控警惕" signal = "bear" # 板块连续大跌提示机会 if sector.recent_gain <= self.sector_dump_opportunity and sector.recent_gain >= -40: suggestion += f"\n✅ 板块累计跌幅{-sector.recent_gain:.1f}%，如果基本面没问题，可考虑适度低吸布局" signal = "bull" if not suggestion: suggestion = "✅ 板块仓位和涨幅正常，无特殊风险" signal = "neutral" return signal, suggestion.strip(), risk_increment class SectorConcentrationRisk: """板块和风格集中度风控""" def __init__(self, max_single_sector: float = 0.25, # 单板块最大仓位25% max_single_style: float = 0.40, # 单风格最大仓位40% max_hot_sectors_total: float = 0.50): # 所有热点板块合计最大50% self.max_single_sector = max_single_sector self.max_single_style = max_single_style self.max_hot_sectors_total = max_hot_sectors_total def check_concentration(self, sectors: List[SectorInfo]) -> Tuple[int, List[str]]: """检查集中度风险，返回总风险增量和警告列表""" risk_increment = 0 warnings = [] # 1. 单板块检查 for sector in sectors: if sector.position_ratio > self.max_single_sector: extra = (sector.position_ratio - self.max_single_sector) * 100 risk_increment += int(extra) warnings.append(f"⚠️ 板块【{sector.name}】仓位{sector.position_ratio:.1%}，超过最大限制{self.max_single_sector:.1%}") # 2. 按风格汇总检查 style_summary: Dict[str, float] = {} for sector in sectors: if sector.style not in style_summary: style_summary[sector.style] = 0.0 style_summary[sector.style] += sector.position_ratio for style, ratio in style_summary.items(): if ratio > self.max_single_style: extra = (ratio - self.max_single_style) * 50 risk_increment += int(extra) warnings.append(f"⚠️ 风格【{style}】总仓位{ratio:.1%}，超过最大限制{self.max_single_style:.1%}，建议分散") # 3. 热点板块合计检查 hot_total = sum(s.position_ratio for s in sectors if s.is_hot) if hot_total > self.max_hot_sectors_total: extra = (hot_total - self.max_hot_sectors_total) * 80 risk_increment += int(extra) warnings.append(f"⚠️ 所有热点板块合计仓位{hot_total:.1%}，超过{self.max_hot_sectors_total:.1%}，总体过热风险") return risk_increment, warnings class HotSectorNewsRiskAdjust: """热点板块消息风险调整，热点更容易利好出尽，灵敏度提高""" def __init__(self, hot_risk_multiplier: float = 2.0, # 热点板块风险分放大倍数 medium_risk_multiplier: float = 1.5): # 中度上涨板块放大倍数 self.hot_risk_multiplier = hot_risk_multiplier self.medium_risk_multiplier = medium_risk_multiplier def adjust_risk_score(self, base_score: int, sector_recent_gain: float) -> int: """根据板块涨幅调整风险分""" if sector_recent_gain >= 50: return int(base_score * self.hot_risk_multiplier) elif sector_recent_gain >= 30: return int(base_score * self.medium_risk_multiplier) else: return base_score def get_structural_risk_report(sectors: List[SectorInfo], timing: StructuredMarketTiming, concentration: SectorConcentrationRisk) -> str: """生成结构化行情风险报告""" timing = StructuredMarketTiming() concentration = SectorConcentrationRisk() total_risk = 0 all_warnings = [] all_suggestions = [] # 逐个板块择时 for sector in sectors: signal, suggestion, risk_inc = timing.timing_sector(sector) total_risk += risk_inc if signal != "neutral": all_suggestions.append(f"【{sector.name}】{suggestion}") # 集中度检查 risk_inc, warnings = concentration.check_concentration(sectors) total_risk += risk_inc all_warnings.extend(warnings) # 生成报告 lines = [] lines.append("=" * 60) lines.append("结构化行情板块风控择时报告") lines.append("=" * 60) lines.append(f"监控板块数量: {len(sectors)}") lines.append(f"总风险增量评分: {total_risk}") lines.append("") if all_warnings: lines.append("⚠️ 集中度风险警告:") for w in all_warnings: lines.append(f" • {w}") lines.append("") if all_suggestions: lines.append("💡 择时建议:") for s in all_suggestions: lines.append(f" • {s}") lines.append("") if total_risk >= 30: lines.append("⚠️ 整体结论：板块风险较高，建议减仓分散") elif total_risk >= 10: lines.append("🟡 整体结论：存在一定板块风险，建议密切观察") else: lines.append("✅ 整体结论：板块结构正常，无明显风险") lines.append("=" * 60) return "\n".join(lines) if __name__ == "__main__": # 测试 print("=== 测试结构化行情风控 ===\n") sectors = [ SectorInfo( name="AI算力", style="AI", recent_gain=65.0, position_ratio=0.22, stock_count=4, is_hot=True ), SectorInfo( name="新能源", style="新能源", recent_gain=-25.0, position_ratio=0.10, stock_count=2, is_hot=False ), SectorInfo( name="消费", style="消费", recent_gain=8.0, position_ratio=0.12, stock_count=3, is_hot=False ) ] timing = StructuredMarketTiming() concentration = SectorConcentrationRisk() print(get_structural_risk_report(sectors, timing, concentration)) # 测试热点风险放大 adjust = HotSectorNewsRiskAdjust() base_score = 10 print(f"\n基础风险分10，不同涨幅放大后:") print(f" 涨幅10% → {adjust.adjust_risk_score(10, 10)}") print(f" 涨幅35% → {adjust.adjust_risk_score(10, 35)}") print(f" 涨幅60% → {adjust.adjust_risk_score(10, 60)}")