知微
123 lines
4.0 KiB
Python
Executable File
123 lines
4.0 KiB
Python
Executable File
#!/usr/bin/env python3
|
||
"""
|
||
prune_branches.py — 每日剪枝
|
||
|
||
扫描所有 strategy_tree 分支，删除低效分支：
|
||
- 触发 >= 3次 且 成功率 < 30% → 标记 pruning_candidate
|
||
- 触发 >= 5次 且 成功率 < 50% → 标记 pruning_candidate
|
||
- pruning_candidate 连续7天无新触发 → 删除
|
||
|
||
自成长核心：低效分支被淘汰，高效分支被保留。
|
||
数据写入 decisions.json 的 strategy_tree.branches[]。
|
||
"""
|
||
|
||
import json, sys, os
|
||
from datetime import datetime, timedelta
|
||
from mo_data import read_decisions
|
||
from mofin_db import get_conn, write_holding_strategy
|
||
|
||
DECISIONS_PATH = "/home/hmo/web-dashboard/data/decisions.json"
|
||
PRUNE_LOG = "/home/hmo/MoFin/data/prune_log.json"
|
||
|
||
|
||
def load_decisions():
|
||
return read_decisions()
|
||
|
||
|
||
def save_decisions(data):
|
||
# DB 写入（替代 json.dump）
|
||
try:
|
||
conn = get_conn()
|
||
for d in data.get("decisions", []):
|
||
write_holding_strategy(conn, d.get("code", ""), d.get("name", ""), d)
|
||
conn.close()
|
||
except Exception:
|
||
pass
|
||
# [migrated to DB] — cold backup removed
|
||
# with open(DECISIONS_PATH, "w") as f:
|
||
# json.dump(data, f, indent=2, ensure_ascii=False)
|
||
|
||
|
||
def main():
|
||
data = load_decisions()
|
||
decisions = data.get("decisions", [])
|
||
today = datetime.now().strftime("%Y-%m-%d")
|
||
pruned = []
|
||
warnings = []
|
||
|
||
for entry in decisions:
|
||
code = entry.get("code", "")
|
||
tree = entry.get("strategy_tree", {})
|
||
branches = tree.get("branches", [])
|
||
if not branches:
|
||
continue
|
||
|
||
keep = []
|
||
for br in branches:
|
||
triggers = br.get("trigger_count", 0)
|
||
success = br.get("success_rate")
|
||
last = br.get("last_triggered", "")
|
||
priority = br.get("priority", 99)
|
||
|
||
# 跳过默认持有分支
|
||
if priority == 99:
|
||
keep.append(br)
|
||
continue
|
||
|
||
# 评估是否该剪枝
|
||
should_prune = False
|
||
reason = ""
|
||
|
||
if triggers >= 5 and success is not None and success < 50:
|
||
should_prune = True
|
||
reason = f"触发{triggers}次，成功率{success}% < 50%"
|
||
elif triggers >= 3 and success is not None and success < 30:
|
||
should_prune = True
|
||
reason = f"触发{triggers}次，成功率{success}% < 30%"
|
||
|
||
if should_prune:
|
||
pruned.append({
|
||
"code": code,
|
||
"branch_id": br.get("id", ""),
|
||
"action": br.get("action", {}).get("type", ""),
|
||
"rationale": br.get("rationale", ""),
|
||
"triggers": triggers,
|
||
"success_rate": success,
|
||
"reason": reason,
|
||
"pruned_at": today,
|
||
})
|
||
print(f"[PRUNE] {code} {br.get('id','?')}: {reason}")
|
||
else:
|
||
keep.append(br)
|
||
|
||
if len(keep) < len(branches):
|
||
tree["branches"] = keep
|
||
entry["strategy_tree"] = tree
|
||
|
||
if pruned:
|
||
save_decisions(data)
|
||
# 记录剪枝日志
|
||
log = []
|
||
try:
|
||
with open(PRUNE_LOG) as f:
|
||
log = json.load(f)
|
||
except Exception:
|
||
pass
|
||
log.append({
|
||
"date": today,
|
||
"pruned": pruned,
|
||
"total_before": sum(len(e.get("strategy_tree", {}).get("branches", [])) for e in decisions),
|
||
})
|
||
os.makedirs(os.path.dirname(PRUNE_LOG), exist_ok=True)
|
||
with open(PRUNE_LOG, "w") as f:
|
||
json.dump(log, f, indent=2, ensure_ascii=False)
|
||
print(f"[PRUNE] 今日剪枝{len(pruned)}条，保留{sum(len(e.get('strategy_tree',{}).get('branches',[])) for e in decisions)}条")
|
||
else:
|
||
print("[PRUNE] 无需要剪枝的分支")
|
||
|
||
return 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
sys.exit(main())
|