自成长:分支评估+剪枝+报告接入

补齐「顺势而为 环境预判 策略多分枝」体系中缺失的组件:

branch_evaluator.py(新增)— 每30min评估所有策略树分支
  1. detect_scenario() 获取当前宏观情景
  2. 对42只股票评估哪个分支当前适用
  3. 适用分支 trigger_count+1, last_triggered=now
  4. 触发>=3次且成功率<30%→标记pruning_candidate
  5. 无决策树的股票自动初始化(init_default_branches)

prune_branches.py(新增)— 每日16:30收盘后剪枝
  阈值:触发>=3次且成功率<30%→裁掉并归档到pruned_branches
  Dad说「每周太低频」→改为每日

stale_push_wlin.py(修改)— 报告每只股增加分支行:
  【弱势震荡→buy_dip】价格回调到支撑区,弱势市场低吸

cron更新:
  分支扫描(b809fcabfa5b) → 指向branch_evaluator.py, 每30min
  剪枝(a3697c108c7b) → 指向prune_branches.py, 每日16:30

自成长核心:branch_evaluator 运行时自动发现并初始化无策略树股票,
252个分支已生成, trigger_count已开始累积, 反馈循环上线
This commit is contained in:
知微
2026-06-24 10:24:11 +08:00
parent eb86a9091e
commit 102a64d856
3 changed files with 260 additions and 15 deletions
+77 -15
View File
@@ -1,21 +1,83 @@
#!/usr/bin/env python3
"""
prune_branches.py — 每周六凌晨6点执行
prune_branches.py — 分支剪枝引擎(每日)
低效分支剪枝
- 触发≥5次且成功率<30% → 剪枝归档
- 输出报告(stdout → cron delivery
裁掉低效分支:trigger_count ≥ 3 且 success_rate < 30%
被剪的分支从 strategy_tree.branches 移除,归档到 strategy_tree.pruned_branches
Dad说"每周"太低频 → 改为每日16:30(收盘后)
"""
import sys, json
sys.path.insert(0, '/home/hmo/MoFin')
from strategy_tree import prune_low_performance_branches
result = prune_low_performance_branches(min_triggers=5, min_success_rate=0.3)
import json, sys
from datetime import datetime
if result:
print(f"[分支剪枝] 已剪枝 {len(result)} 条低效分支:")
for item in result:
print(f"{item}")
print("[分支剪枝] 剪枝完成")
else:
print("[分支剪枝] 无低效分支需剪枝")
DECISIONS_PATH = "/home/hmo/web-dashboard/data/decisions.json"
# 剪枝阈值:触发≥3次且成功率<30%
TRIGGER_MIN = 3
SUCCESS_MAX = 30
def prune():
try:
with open(DECISIONS_PATH) as f:
data = json.load(f)
except Exception as e:
print(f"[错误] 读 decisions.json 失败: {e}", file=sys.stderr)
return 1
decisions = data.get("decisions", [])
total_pruned = 0
results = []
for entry in decisions:
code = entry.get("code", "")
tree = entry.get("strategy_tree")
if not tree:
continue
branches = tree.get("branches", [])
if not branches:
continue
pruned_branches = tree.get("pruned_branches", [])
kept = []
for br in branches:
tc = br.get("trigger_count", 0)
sr = br.get("success_rate")
if tc >= TRIGGER_MIN and sr is not None and sr < SUCCESS_MAX:
# 归档
br["pruned_at"] = datetime.now().isoformat()
pruned_branches.append(br)
total_pruned += 1
results.append({
"code": code,
"branch_id": br.get("id", "?"),
"trigger_count": tc,
"success_rate": sr,
"rationale": br.get("rationale", ""),
})
else:
kept.append(br)
tree["branches"] = kept
tree["pruned_branches"] = pruned_branches
tree["last_pruned"] = datetime.now().isoformat() if total_pruned > 0 else tree.get("last_pruned", "")
with open(DECISIONS_PATH, "w") as f:
json.dump(data, f, indent=2, ensure_ascii=False)
if total_pruned > 0:
lines = [f"【分支剪枝】本次裁掉{total_pruned}个低效分支"]
for r in results:
lines.append(f"{r['code']}/{r['branch_id']}(触发{r['trigger_count']}次/成功率{r['success_rate']}%)")
lines.append(f" 理由: {r['rationale']}")
print("\n".join(lines))
return 0
else:
# 静默
print(f"【分支剪枝】无需剪枝(所有分支均未达到触发{TRIGGER_MIN}次且成功率<{SUCCESS_MAX}%的阈值)")
return 0
if __name__ == "__main__":
sys.exit(prune())