bbqddt2/Antigravity
0
1# -*- coding: utf-8 -*-2"""3Antigravity 策略框架 — 假设验证器 V1.04 5核心理念:6- 每个假设都必须通过 walk-forward 交叉验证7- 验证结果: survived (通过) / eliminated (失败)8- 失败时记录原因: overfit / noise_chase / cliche / fragile9- 验证通过的假设可以"进化"出新假设10 11用法:12 from strategy_proposer.validator import HypothesisValidator13 14 validator = HypothesisValidator()15 results = validator.validate_all(hypotheses, draws)16"""17import json18import math19from pathlib import Path20from typing import Dict, List, Optional, Any21from datetime import datetime22 23from strategy_proposer.hypothesis import Hypothesis24 25_PROJECT_ROOT = Path(__file__).resolve().parent26 27 28class HypothesisValidator:29 """假设验证器 — walk-forward 交叉验证"""30 31 def __init__(self, random_baseline: float = 2.0):32 self.random_baseline = random_baseline33 34 def validate(self, hypothesis: Hypothesis, draws: Any,35 n_windows: int = 8, window_size: int = 300,36 step: int = 30) -> Dict:37 """38 验证一个假设。39 40 由于假设是"陈述"而非具体策略，验证器根据假设的category和params41 生成对应的测试策略，然后进行walk-forward验证。42 43 Args:44 hypothesis: 待验证的假设45 draws: 历史数据46 n_windows: 验证窗口数47 window_size: 训练窗口大小48 step: 窗口步进49 50 Returns:51 {52 "status": "survived" | "eliminated",53 "avg_hits": 平均命中数,54 "p_value": p值,55 "elimination_reason": 失败原因(如果淘汰),56 "per_round": 每轮命中数,57 "validated_at": 验证时间,58 }59 """60 # 根据假设生成测试策略61 strategy = self._build_test_strategy(hypothesis, draws)62 if strategy is None:63 return {64 "status": "eliminated",65 "avg_hits": 0, "p_value": 1.0,66 "elimination_reason": "cannot_build_strategy",67 "per_round": [],68 "validated_at": datetime.now().isoformat(),69 }70 71 # walk-forward 验证72 per_round_hits = []73 for w in range(n_windows):74 train_end = window_size + w * step75 test_start = train_end + 1076 test_end = test_start + 1077 78 if test_end > len(draws):79 break80 81 try:82 pred = strategy(train_draws=draws[:train_end], test_draws=draws[test_start:test_end])83 if pred:84 actual_reds = set()85 for d in pred["test_draws"]:86 actual = d.reds if hasattr(d, 'reds') else sorted(list(d.red))87 actual_reds.update(actual)88 89 hits = len(set(pred.get("reds", [])) & actual_reds) if pred.get("reds") else 090 per_round_hits.append(hits / 6.0)91 else:92 per_round_hits.append(0)93 except Exception:94 per_round_hits.append(0)95 96 if not per_round_hits:97 return {98 "status": "eliminated",99 "avg_hits": 0, "p_value": 1.0,100 "elimination_reason": "no_valid_rounds",101 "per_round": [],102 "validated_at": datetime.now().isoformat(),103 }104 105 avg = sum(per_round_hits) / len(per_round_hits)106 std = math.sqrt(sum((h - avg) ** 2 for h in per_round_hits) / max(len(per_round_hits) - 1, 1))107 108 # t检验109 se = std / math.sqrt(len(per_round_hits)) if std > 0 else 1110 t_stat = (avg - self.random_baseline) / se if se > 0 else 0111 p_value = self._t_to_pvalue(t_stat, len(per_round_hits))112 113 # 判定是否存活114 elimination_reason = None115 if avg < self.random_baseline:116 elimination_reason = "below_random_baseline"117 elif p_value > 0.1:118 elimination_reason = "not_significant"119 elif std > avg * 0.5:120 elimination_reason = "too_unstable"121 elif len(per_round_hits) < 3:122 elimination_reason = "insufficient_data"123 124 status = "eliminated" if elimination_reason else "survived"125 126 # 更新假设127 hypothesis.status = status128 hypothesis.tested_at = datetime.now().isoformat()129 hypothesis.result = {130 "avg_hits": round(avg, 4),131 "std": round(std, 4),132 "p_value": round(p_value, 4),133 "rounds": len(per_round_hits),134 "beats_random": avg > self.random_baseline,135 "stable": round(avg - std, 4),136 }137 hypothesis.elimination_reason = elimination_reason138 139 return {140 "status": status,141 "avg_hits": round(avg, 4),142 "std": round(std, 4),143 "p_value": round(p_value, 4),144 "elimination_reason": elimination_reason,145 "per_round": [round(h, 4) for h in per_round_hits],146 "validated_at": hypothesis.tested_at,147 }148 149 def validate_all(self, hypotheses: List[Hypothesis], draws: Any,150 **kwargs) -> Dict[str, Dict]:151 """批量验证假设"""152 results = {}153 for h in hypotheses:154 result = self.validate(h, draws, **kwargs)155 results[h.id] = result156 return results157 158 def _build_test_strategy(self, hypothesis: Hypothesis, draws: Any):159 """160 根据假设生成测试策略。161 162 这是一个简化实现。在实际系统中，每个假设类别应该有对应的策略生成器。163 这里我们用启发式方法根据假设的category生成简单的测试策略。164 """165 category = hypothesis.template_category166 167 if category in ["periodic", "spectral"]:168 return self._periodic_test_strategy169 elif category == "omission":170 return self._omission_test_strategy171 elif category == "positional":172 return self._positional_test_strategy173 elif category == "cooccurrence":174 return self._cooccurrence_test_strategy175 elif category == "structural":176 return self._structural_test_strategy177 elif category == "blue":178 return self._blue_test_strategy179 else:180 return self._generic_test_strategy181 182 def _periodic_test_strategy(self, train_draws, test_draws):183 """周期性测试策略"""184 from collections import Counter185 freq = Counter()186 for d in train_draws:187 reds = d.reds if hasattr(d, 'reds') else sorted(list(d.red))188 freq.update(reds)189 top6 = [n for n, _ in freq.most_common(6)]190 return {"reds": top6, "test_draws": test_draws}191 192 def _omission_test_strategy(self, train_draws, test_draws):193 """遗漏测试策略"""194 from collections import Counter195 freq = Counter()196 for d in train_draws:197 reds = d.reds if hasattr(d, 'reds') else sorted(list(d.red))198 freq.update(reds)199 # 选频率最低的6个200 sorted_nums = sorted(freq.items(), key=lambda x: x[1])201 top6 = [n for n, _ in sorted_nums[:6]]202 return {"reds": top6, "test_draws": test_draws}203 204 def _positional_test_strategy(self, train_draws, test_draws):205 """位置测试策略"""206 from collections import Counter207 pos_freq = [Counter() for _ in range(6)]208 for d in train_draws:209 reds = d.reds if hasattr(d, 'reds') else sorted(list(d.red))210 for i, r in enumerate(reds):211 pos_freq[i][r] += 1212 top6 = []213 for i in range(6):214 if pos_freq[i]:215 top6.append(pos_freq[i].most_common(1)[0][0])216 return {"reds": top6[:6], "test_draws": test_draws}217 218 def _cooccurrence_test_strategy(self, train_draws, test_draws):219 """共现测试策略"""220 from collections import Counter221 cooccur = Counter()222 for d in train_draws:223 reds = sorted(d.reds if hasattr(d, 'reds') else sorted(list(d.red)))224 for i in range(len(reds)):225 for j in range(i + 1, len(reds)):226 cooccur[(reds[i], reds[j])] += 1227 # 选共现最高的号码228 num_score = Counter()229 for (a, b), c in cooccur.items():230 num_score[a] += c231 num_score[b] += c232 top6 = [n for n, _ in num_score.most_common(6)]233 return {"reds": top6, "test_draws": test_draws}234 235 def _structural_test_strategy(self, train_draws, test_draws):236 """结构测试策略"""237 from collections import Counter238 freq = Counter()239 for d in train_draws:240 reds = d.reds if hasattr(d, 'reds') else sorted(list(d.red))241 freq.update(reds)242 top6 = [n for n, _ in freq.most_common(6)]243 return {"reds": top6, "test_draws": test_draws}244 245 def _blue_test_strategy(self, train_draws, test_draws):246 """蓝球测试策略"""247 from collections import Counter248 blues = Counter()249 for d in train_draws:250 blues[d.blue if hasattr(d, 'blue') else d.blue] += 1251 top_blue = blues.most_common(1)[0][0] if blues else 8252 return {"reds": [], "blue": top_blue, "test_draws": test_draws}253 254 def _generic_test_strategy(self, train_draws, test_draws):255 """通用测试策略（频率最高）"""256 from collections import Counter257 freq = Counter()258 for d in train_draws:259 reds = d.reds if hasattr(d, 'reds') else sorted(list(d.red))260 freq.update(reds)261 top6 = [n for n, _ in freq.most_common(6)]262 return {"reds": top6, "test_draws": test_draws}263 264 @staticmethod265 def _t_to_pvalue(t: float, df: int) -> float:266 """简化t检验p值"""267 x = abs(t) / math.sqrt(2)268 try:269 sign = 1 if x >= 0 else -1270 a1, a2, a3, a4, a5 = 0.254829592, -0.284496736, 1.421413741, -1.453152027, 1.061405429271 p = 0.3275911272 t_val = 1.0 / (1.0 + p * abs(x))273 erf_x = sign * (1.0 - (((((a5 * t_val + a4) * t_val) + a3) * t_val + a2) * t_val + a1) * t_val * math.exp(-x * x))274 p_one_tail = 0.5 * (1 - erf_x)275 return min(1.0, 2 * p_one_tail)276 except:277 return 1.0278