71 lines
2.7 KiB
Python
71 lines
2.7 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""evolution/precompute_evolution.py — 进化机制预计算（2026-08-16）
|
||
定期（每周/每日）预计算进化机制数据，供研究Tab展示（API 只读快照，不实时重算）：
|
||
1. 假设归纳（方向一）：每个激活策略的归纳优化假设
|
||
2. B组候选（方向二）：由果及因挖掘的候选
|
||
3. 策略资格概览
|
||
输出：data/evolution_center.json
|
||
"""
|
||
import sys, os, json
|
||
from datetime import datetime
|
||
|
||
sys.path.insert(0, "/home/hmo/MoFin")
|
||
sys.path.insert(0, "/home/hmo/MoFin/evolution")
|
||
sys.path.insert(0, "/home/hmo/MoFin/deploy/profile-scripts")
|
||
|
||
OUT = "/home/hmo/MoFin/data/evolution_center.json"
|
||
|
||
|
||
def active_versions():
|
||
try:
|
||
d = json.load(open("/home/hmo/MoFin/data/strategy_weights.json", encoding="utf-8"))
|
||
vs = list(d.get("active") or [])
|
||
vs += list(((d.get("markets") or {}).get("hk") or {}).get("active") or [])
|
||
return list(dict.fromkeys(vs))
|
||
except Exception:
|
||
return []
|
||
|
||
|
||
def main():
|
||
print("=== 进化机制预计算开始 ===", flush=True)
|
||
from hypothesis_miner import induce_hypotheses
|
||
from strategy_qualify import evaluate_all_regimes, get_benchmarks
|
||
|
||
out = {"generated_at": datetime.now().strftime("%Y-%m-%d %H:%M:%S"), "hypotheses": [], "b_group": [], "qual_overview": []}
|
||
|
||
# 1. 假设归纳（方向一）
|
||
for v in active_versions():
|
||
mkt = "hk" if v.startswith("hk") else "a"
|
||
try:
|
||
hs, _ = induce_hypotheses(v, mkt, period_tag="2y")
|
||
for h in hs[:3]:
|
||
out["hypotheses"].append({"strategy": v, "market": mkt, **h})
|
||
print(f" 假设 [{v}]: {h['hypothesis'][:60]}", flush=True)
|
||
except Exception as e:
|
||
print(f" 假设 [{v}] 失败: {e}", flush=True)
|
||
|
||
# 2. B组候选（方向二）
|
||
try:
|
||
p = json.load(open("/home/hmo/MoFin/data/b_group_candidates.json", encoding="utf-8"))
|
||
out["b_group"] = p.get("candidates", [])
|
||
print(f" B组候选: {len(out['b_group'])}", flush=True)
|
||
except Exception as e:
|
||
print(f" B组读取失败: {e}", flush=True)
|
||
|
||
# 3. 资格概览
|
||
for v in active_versions():
|
||
mkt = "hk" if v.startswith("hk") else "a"
|
||
try:
|
||
q = evaluate_all_regimes(v, mkt, bench=get_benchmarks(mkt))
|
||
out["qual_overview"].append({"strategy": v, "market": mkt, "qualification": q})
|
||
except Exception:
|
||
pass
|
||
|
||
with open(OUT, "w", encoding="utf-8") as f:
|
||
json.dump(out, f, ensure_ascii=False, indent=1)
|
||
print(f"写入 {OUT}: hypotheses={len(out['hypotheses'])} b_group={len(out['b_group'])} qual={len(out['qual_overview'])}")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|