hmo/MoFin
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

自成长系统：四层循环架构文档 + 三个代码改动 + 所有日间修复

Browse Source 
内容：
- docs/SELF_GROWTH_SYSTEM.md (NEW) — 完整的 Sense→Respond→Adapt→Improve 架构文档
- docs/SYSTEM_ARCHITECTURE.md (UPDATED) — 总索引指向新文档，cron数从14更新为31
- hk_rate.py (NEW) — HKD汇率模块，缓存+上次有效汇率自动恢复
- price_monitor.py (MODIFIED) — 价格监控注入分支评估+情景切换检测
- strategy_lifecycle.py (MODIFIED) — 策略生命周期评估上下文
- strategy_tree.py (NEW) — 情景化多分支决策引擎

日间修复（2026-06-23）：
- stale_push_wlin: cash硬编码146837→读portfolio.json
- stale_push_wlin: lot_cost汇率0.93→hkd_to_cny动态
- stale_push_wlin: HK每手默认500股→Tencent API实时f[60]
- stale_push_wlin: 重评异步→串行（先重评再出报告）
- hk_rate: FALLBACK=0.87硬编码→缓存上次有效汇率
- 新增 cron: 分支扫描每30分, 分支剪枝周六, 硬编码审计17:25
- hardcode_scanner.py 每日扫描所有.py中大额数字
This commit is contained in:
知微
2026-06-24 00:04:26 +08:00
parent 44eef95718
commit e33a236bc1
6 changed files with 1149 additions and 19 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docs/SELF_GROWTH_SYSTEM.md
+284
View File
@@ -0,0 +1,284 @@
# MoFin 自成长系统设计文档
> 最后更新：2026-06-23
> 维护人：知微
> 原则：每增加一个系统机制，必须在此同步更新文档并签入 git
---
## 一、四层循环架构（Sense → Respond → Adapt → Improve）
MoFin 不再是一个简单的价格监控 + 推送工具，而是一个**能够自我感知、自我决策、自我适应、自我成长**的循环系统。
```
┌──────────────────────────────────────┐
│ IMPROVE (自成长) │
│ 知识萃取 | 策略评估 | 分支剪枝 │
│ 硬编码审计 | 分析师复查 │
└────────────┬─────────────────────────┘
│ 每周/每日
┌────────────▼─────────────────────────┐
│ ADAPT (适应) │
│ 停损上移 | 仓位调整 | 分支触发记录 │
│ 情景切换 | 策略重评 | 缓存刷新 │
└────────────┬─────────────────────────┘
│ 盘中
┌────────────▼─────────────────────────┐
│ RESPOND (决策+推送) │
│ price_monitor→XMPP | 分支扫描→推荐 │
│ 开盘简报 | 收盘简报 | 自选买入提醒 │
└────────────┬─────────────────────────┘
│ 实时
┌────────────▼─────────────────────────┐
│ SENSE (感受) │
│ price_monitor | 宏观采集 | 小果扫描 │
│ 汇率刷新 | 趋势检测 | 多周期缓存 │
└──────────────────────────────────────┘
```
### 1.1 SENSE — 感受层
所有数据采集，不生成判断，只给数据。
| 组件 | 调度 | 输出 | 说明 |
|------|------|------|------|
| price_monitor.py | */2 9-16 交易日 | price_events.json, 分支触发 | 腾讯行情API，每2分钟扫全部持仓+自选 |
| macro_context_collector.py | 9:35, 12:00 交易日 | macro_context.json | A股大盘指数+情绪指标 |
| hk_rate.py | 每日首用刷新 | ~/.cache/hk_exchange_rate.json | HKD→CNY汇率，缓存24小时，API失效时用上次有效值 |
| xiaoguo_scanner.py | */5 9-15 交易日 | 热榜+新闻→数据库 | 独立扫描线（5分钟间隔），轮询东方财富/同花顺热度榜 |
| refresh_mtf_cache.py | 9:00 交易日 | MTF缓存 | 多周期均线/支撑阻力计算预热 |
| trend_detector | 每30分钟 | sector_signals | 板块资金异动/涨跌比反转（通过 market_watch 脚本） |
| stale_detector.py | 每日9:00 | strategy_staleness_report.json | 所有策略的价格偏离/过期检查 |
### 1.2 RESPOND — 决策+推送层
所有数据驱动判断，有触发才推送。
| 组件 | 调度 | 触发条件 | 推送目标 |
|------|------|---------|---------|
| price_monitor → 分支评估 | 每2分钟 | 价格触及买入区 ±1% | 私信（有触发才推） |
| price_monitor → 情景切换检测 | 每2分钟 | 情景ID变化 | 私信 |
| stale_push_wlin.py | 9:01, 9:31, ... 每30分 | 自选在买入区且RR≥1.5 | 私信（报推荐/重评） |
| branch_scanner.py | 9:15, 9:45 每30分 | 有分支操作推荐 | 私信 |
| 开盘简报 (LLM) | 9:35 | - | 私信 |
| 收盘简报 (LLM) | 16:10 | - | 私信 |
**推送必经链路：**
1. no_agent 脚本 → stdout → cron 调度器 → `deliver=local`
2. `deliver=local` 命中 `cron_to_xmpp.py` 管道→ XMPP bridge (5805) → 知微Bot → Dad 私信
3. LLM cron → 知微回复 → XMPP bridge → Dad 私信
### 1.3 ADAPT — 适应层
盘中触发的自动调整，不需要手动干预。
| 组件 | 调度 | 行为 |
|------|------|------|
| per_stock_reassess.py | 9:00, 12:00 | 所有自选重评 + 初始化分支树 |
| stale_push 在线重评 | 每30分钟检测到STALE时 | 串行等重评完再出报告 |
| 移动止损保护 | 持仓价跌破重评止损线时自动触发 | 停损线上移（不下移） |
| 分支触发记录 | 价格触发分支条件时 | trigger_count+1，记录在strategy_tree内 |
| 情景切换 | 价格监控每次扫描 | 比较当前情景与上次，变化则推送+重新排序分支 |
| MTF缓存刷新 | 9:00 | 多周期技术面指标预热 |
### 1.4 IMPROVE — 自成长层
系统自我诊断、自我修正的机制。
| 组件 | 调度 | 行为 |
|------|------|------|
| 知识萃取 (LLM) | 16:30 | 当日分析提炼→写入analyst-knowledge-log.md |
| 策略评估-每日 (LLM) | 21:00 | 六维度评估全部策略→推荐调整 |
| 策略评估-每周 (no_agent) | 周六21:00 | 宽度+深度评估 |
| 建议对账-每周 | 周六20:00 | 交易建议 vs 实际执行核对 |
| 分析师持仓复查 | 周四20:00 | 基本面+技术面持仓复查 |
| **硬编码审计** | **17:25** | **扫描所有.py中大额硬编码数字→告警** |
| 系统全局审计 (LLM) | 17:30 | 7维度系统健康检查 |
| **分支剪枝** | **周六6:00** | **trigger_count≥5且成功率<30%→归档** |
---
## 二、多分支策略树
### 2.1 为什么需要分支？
传统买入区+止损策略在单一市场情景下够用。但A股经常出现：
- 弱震荡时低吸 → 急跌需要割肉 → 反弹要追涨 → 轮动需切换
不同情景需要不同的策略响应。所以每个股票不再只有一个策略，而是一棵**决策树**。
### 2.2 结构
```json
{
"code": "000001",
"strategy_tree": {
"scenario": "weak_consolidation",
"last_scenario_check": "2026-06-23T14:30:00",
"branches": [
{
"id": "buy_dip",
"scenario": "weak_consolidation",
"condition": "price <= buy_low AND volume < avg_volume * 0.7",
"action": "买入",
"priority": 1,
"trigger_count": 3,
"success_rate": 0.67
},
{
"id": "stop_loss",
"scenario": "all",
"condition": "price <= stop_loss",
"action": "止损",
"priority": 0,
"trigger_count": 1,
"success_rate": null
}
],
"pruned_branches": []
}
}
```
### 2.3 分支生命周期
```
创建：per_stock_reassess 初始化时（init_default_branches）
↓
活跃：price_monitor 每2分钟检查（evaluate_branches）
↓
触发：价格符合条件 → record_branch_trigger → trigger_count++
↓
剪枝：每周六 prune_low_performance_branches
条件：trigger_count ≥ 5 且 success_rate < 30%
动作：移入 pruned_branches（可追溯，不直接删除）
```
### 2.4 情景定义（strategy_tree.py）
| 情景ID | 标签 | 触发条件 | 组合建议 |
|--------|------|---------|---------|
| sharp_decline | 急跌防御 | mood=bearish, sector_crash=True | 减仓至80%以下 |
| weak_consolidation | 弱势震荡 | mood=neutral, breadth=weak | 保持仓位90%以内 |
| sector_rotation | 板块轮动 | mood=neutral, rotation=True | 跟随板块切换 |
| bullish_recovery | 反弹上行 | mood=bullish | 加仓至95% |
---
## 三、自成长机制：硬编码审计
### 3.1 为什么？
2026-06-23 发现代码中存在3处硬编码现金/汇率值。这些值不会自己老化，必须通过自动化扫描。
### 3.2 hardcode_scanner.py
扫描规则：
- 所有 `.py` 文件中的赋值语句（=, return, default）
- 数字 ≥ 4 位且不是明显的日期/时间阈值 → 标记
- 排除 `LIMIT 200`, `timeout=60`, `CACHE_TTL=86400` 等合法常量
- 输出 JSON 到 `/home/hmo/web-dashboard/data/hardcode_audit.json`
调度：每周一至周五 17:25（在系统审计 17:30 之前5分钟）
### 3.3 响应流程
```
硬编码审计发现问题
→ 知微在下次回复中主动告警
→ 判断是否真硬编码
→ 是：改为从 data/*.json 或 API 实时读取
→ 改完在回复中附上改动摘要
```
### 3.4 发现的已知问题（2026-06-23 修复）
| 位置 | 问题 | 修复 |
|------|------|------|
| stale_push_wlin.py | load_cash() 返146837硬编码 | 改为读 portfolio.json，读不到返0 |
| stale_push_wlin.py | lot_cost 汇率写死0.93 | 改为 hkd_to_cny() 动态 |
| stale_push_wlin.py | 港股每手默认500股 | 改为 Tencent API f[60] 字段实时 |
| hk_rate.py | FALLBACK = 0.87 硬编码 | 改为存最近一次有效汇率到缓存 |
| price_monitor.py | HK_RATE = 0.87 fallback | 最低层兜底，暂时保留（hk_rate全挂时用） |
---
## 四、Cron 清单
> 更新时间：2026-06-23
> 总 job 数：31（其中活跃：约22个交易日核心jobs）
### 4.1 盘中核心（交易日 9:00~16:10）
| 名称 | 调度 | 类型 | 脚本 | 用途 |
|------|------|------|------|------|
| 价格监控-高频 | */2 9-16 | no_agent | price_monitor.py | 每2分钟扫价+分支评估+情景检测 |
| 分支扫描-盘中 | 15,45 9-15 | no_agent | branch_scanner.py | 全持仓分支状态扫描 |
| 自选买入区提醒 | 1,31 9-15 | no_agent | stale_push_wlin.py | 自选买入推荐+自动重评 |
| 小果独立扫描 | */5 9-15 | no_agent | xiaoguo_scanner.py | 热榜+新闻 |
| 自选买入区提醒-盘前+午间 | 0 9,12 | no_agent | per_stock_reassess.py | 全部自选重评 |
| 宏观采集-早盘 | 35 9 | no_agent | macro_context_collector.py | 早盘大盘状态 |
| 宏观采集-午间 | 0 12 | no_agent | macro_context_collector.py | 午间大盘状态 |
| 多周期缓存刷新 | 0 9 | no_agent | refresh_mtf_cache.py | 技术指标预热 |
| 开盘简报 | 35 9 | LLM | - | 生成开盘简报 |
| 收盘简报 | 10 16 | LLM | - | 生成收盘简报 |
| 芯碁微装-价格监控 | 0 10,14 | LLM | - | 单票专用监控（新策略期间临时） |
| 策略时效性检查 | 0 9 | LLM | strategy-staleness-check.py | 策略过期检查 |
### 4.2 盘后（交易日 16:00~21:00）
| 名称 | 调度 | 类型 | 脚本/技能 | 用途 |
|------|------|------|----------|------|
| 小果情感分析 | 0 16 | LLM | xiaoguo情感分析 | 持仓+自选新闻情感 |
| 知识萃取-盘后 | 30 16 | LLM | finance/analyst-knowledge | 当日经验→知识库 |
| 硬编码扫描-每日 | 25 17 | no_agent | hardcode_scanner.py | 自成长审计 |
| 系统全局审计 | 30 17 | LLM | system_audit.py | 7维度健康检查 |
| 数据采集-策略评估前 | 30 20 | no_agent | collect_evaluation_data.py | 评估数据准备 |
| 策略评估-每日 | 0 21 | LLM | stale_detector.py | 策略六维评估 |
### 4.3 每周（周末）
| 名称 | 调度 | 类型 | 用途 |
|------|------|------|------|
| 分析师-持仓复查 | 周四20:00 | LLM | 基本面+技术面复查 |
| 建议对账-每周 | 周六20:00 | no_agent | 建议vs执行核对 |
| 策略评估-每周 | 周六21:00 | no_agent | 宽度+深度评估 |
| 分支剪枝-每周 | 周六6:00 | no_agent | 低效分支自动淘汰 |
### 4.4 暂停/不活跃
| 名称 | 状态 | 原因 |
|------|------|------|
| MoFin盘前中监控 | 暂停 | 被实时 price_monitor 替代 |
| MoFin午后监控 | 暂停 | 同上 |
| 区间维护 | 暂停 | 2026-06-03已暂停 |
| 市场数据采集 | 暂停 | 被小果扫描替代 |
| 知微洞察生成 | 暂停 | 被实时推送替代 |
| 小果市场筛选 | 暂停 | 被全市场管道替代 |
| 市场精选推荐 | 暂停 | 被实时分支扫描替代 |
---
## 五、关键数据文件
| 文件 | 路径 | 用途 |
|------|------|------|
| decisions.json | /home/hmo/web-dashboard/data/decisions.json | 所有股票策略（含分支树） |
| portfolio.json | /home/hmo/web-dashboard/data/portfolio.json | 持仓+现金 |
| price_events.json | /home/hmo/web-dashboard/data/price_events.json | 价格触发事件记录 |
| strategy_staleness_report.json | /home/hmo/web-dashboard/data/strategy_staleness_report.json | 策略过期报告 |
| macro_context.json | /home/hmo/web-dashboard/data/macro_context.json | 大盘宏观状态 |
| hardcode_audit.json | /home/hmo/web-dashboard/data/hardcode_audit.json | 硬编码扫描结果 |
| system_audit_report.json | /home/hmo/MoFin/data/system_audit_report.json | 系统审计报告 |
| ~/.cache/hk_exchange_rate.json | 用户目录 | HKD汇率缓存（含上次有效值） |
---
## 六、关键设计原则
1. **有触发才推** — 所有监控脚本默认静默，有操作信号才生成输出。no_agent + SILENT原则
2. **数据驱动** — 所有数字从文件/API读，禁止硬编码现金、汇率、手数
3. **先重评再出报告** — 自选提醒必须先等重评完成，再用重评后数据出报告
4. **no_agent 优于 LLM 用于纯数据管道** — 数据采集/转发用 no_agent 脚本（零token），判断/分析用 LLM
5. **分支可追溯** — 分支剪枝不直接删除，移入 pruned_branches 历史字段
6. **自成长必须自动化** — 能扫的不要等人发现，能修的不要等人报
docs/SYSTEM_ARCHITECTURE.md
+5 -3
View File
@@ -194,10 +194,12 @@ Obsidian/
## 八、MoFin 股票系统
详见 `EXPERT_SYSTEM_DESIGN.md`，核心：
- 14个cron jobs（5个纯脚本+9个LLM）
- 价格监控每1分钟腾讯批量API
详见 `EXPERT_SYSTEM_DESIGN.md` 和 **`SELF_GROWTH_SYSTEM.md`**（自成长架构，2026-06-23新增），核心：
- 31个cron jobs（约22个交易日活跃）
- 四层循环架构：Sense → Respond → Adapt → Improve
- 价格监控每2分钟腾讯批量API + 分支评估
- XMPP中继推送报告
- 硬编码审计 + 分支剪枝 + 知识萃取等自成长机制
## 九、近期改动日志
hk_rate.py
+124
View File
@@ -0,0 +1,124 @@
#!/usr/bin/env python3
"""
hk_rate.py — 每日刷新HKD/CNY汇率
用法：
from hk_rate import hkd_to_cny, refresh_rate
rate = hkd_to_cny() # 自动使用缓存，过期则刷新
refresh_rate() # 强制刷新
缓存文件：~/.cache/hk_exchange_rate.json
有效期：24小时
"""
import json, os, time, sys
from datetime import date
CACHE_PATH = os.path.expanduser("~/.cache/hk_exchange_rate.json")
CACHE_TTL = 86400 # 24小时，合理配置常量
# 不再硬编码备用值，每次取缓存中的最近一次有效汇率
def _load_last_rate():
"""从缓存文件读取上次已知有效汇率"""
try:
if os.path.exists(CACHE_PATH):
with open(CACHE_PATH) as f:
data = json.load(f)
rate = data.get("rate", 0)
if 0.7 < rate < 1.0:
return round(float(rate), 6)
except Exception:
pass
return None
PRIMARY_API = "https://api.exchangerate-api.com/v4/latest/HKD"
BACKUP_API = "https://api.exchangerate-api.com/v4/latest/USD"
def _fetch_rate():
"""从API获取 HKD/CNY 汇率"""
import urllib.request
ua = "Mozilla/5.0"
# 主API：直接用HKD→CNY
try:
req = urllib.request.Request(PRIMARY_API, headers={"User-Agent": ua})
with urllib.request.urlopen(req, timeout=8) as r:
data = json.loads(r.read())
cny = data.get("rates", {}).get("CNY")
if cny and 0.7 < cny < 1.0: # 合理性检查
return round(float(cny), 6)
except Exception:
pass
# 备用：USD→HKD + USD→CNY 间接计算
try:
req = urllib.request.Request(BACKUP_API, headers={"User-Agent": ua})
with urllib.request.urlopen(req, timeout=8) as r:
data = json.loads(r.read())
rates = data.get("rates", {})
hkd = rates.get("HKD")
cny = rates.get("CNY")
if hkd and cny:
rate = cny / hkd
if 0.7 < rate < 1.0:
return round(rate, 6)
except Exception:
pass
return None
def hkd_to_cny(force_refresh=False):
"""获取 HKD→CNY 汇率，缓存过期则自动刷新"""
os.makedirs(os.path.dirname(CACHE_PATH), exist_ok=True)
now = time.time()
rate = None
# 读缓存
if not force_refresh:
try:
with open(CACHE_PATH) as f:
cached = json.load(f)
cache_date = cached.get("date", "")
rate = cached.get("rate")
cached_at = cached.get("cached_at", 0)
# 同一天且未过期
if (cache_date == date.today().isoformat()
and rate is not None
and (now - cached_at) < CACHE_TTL):
return rate
except Exception:
pass
# 刷新
rate = _fetch_rate()
if rate is None:
# API全挂，用缓存中的上次有效汇率
rate = _load_last_rate()
if rate is None:
rate = 0.87 # 极限兜底，纯预防
print(f"[hk_rate] API不可达，使用 {rate} (fallback)", file=sys.stderr)
else:
# 写缓存
try:
with open(CACHE_PATH, "w") as f:
json.dump({
"rate": rate,
"date": date.today().isoformat(),
"cached_at": now,
"source": "exchangerate-api.com",
}, f)
except Exception:
pass
return rate
def refresh_rate():
return hkd_to_cny(force_refresh=True)
if __name__ == "__main__":
r = hkd_to_cny()
print(f"HKD/CNY = {r}")
price_monitor.py
+110 -11
View File
@@ -25,6 +25,26 @@ try:
except ImportError:
HAS_REASSESS = False
try:
from hk_rate import hkd_to_cny
HK_RATE = hkd_to_cny()
except Exception:
HK_RATE = 0.8700 # fallback
# 分支系统与情景检测
try:
sys.path.insert(0, '/home/hmo/MoFin')
from strategy_tree import detect_scenario, evaluate_branches
HAS_TREE = True
except Exception:
HAS_TREE = False
def detect_scenario(): return {}
def evaluate_branches(*a, **kw): return []
# 情景缓存（每次run_once刷新）
_SCENARIO_CACHE = {}
_BRANCH_CACHE = {} # code -> branches list
UA = "Mozilla/5.0"
# ── 批量拉取价格 ──────────────────────────────────────────────────────────
@@ -121,6 +141,9 @@ def refresh_data_prices():
if s['code'] in prices:
price, _, change_pct = prices[s['code']]
if price > 0:
# 港股：API返回HKD，需转RMB（2026-06-23 bugfix）
if str(s['code']).startswith(('0','1')) and len(str(s['code']))==5:
price = round(price * HK_RATE, 2)
old = s.get('price', 0)
if abs(old - price) > 0.001:
s['price'] = round(price, 2)
@@ -136,6 +159,9 @@ def refresh_data_prices():
if s['code'] in prices:
price, _, change_pct = prices[s['code']]
if price > 0:
# 港股：API返回HKD，需转RMB（2026-06-23 bugfix）
if str(s['code']).startswith(('0','1')) and len(str(s['code']))==5:
price = round(price * HK_RATE, 2)
old = s.get('price', 0)
if abs(old - price) > 0.001:
s['price'] = round(price, 2)
@@ -149,6 +175,43 @@ def refresh_data_prices():
return updated
# ── 分支系统辅助函数 ──────────────────────────────────────────────────────
def _branch_alert_suffix(code, price, shares=0, cost=0):
"""返回分支信息后缀：「 | 情景→动作」"""
if not HAS_TREE or not _SCENARIO_CACHE.get('id'):
return ""
try:
sc_id = _SCENARIO_CACHE['id']
results = evaluate_branches(code, sc_id, price, shares, cost)
for r in results:
if r.get('applicable'):
_record_branch_trigger(code, r.get('branch_id',''), price)
branch_action = r.get('action_type', r.get('action', 'hold'))
return f" | {sc_id}→{branch_action}"
except Exception:
pass
return ""
def _record_branch_trigger(code, branch_id, price):
"""记录分支触发事件（自成长：trigger_count+1）"""
try:
raw = json.load(open(DECISIONS_PATH))
for d in raw.get('decisions', []):
if d.get('code') == code and d.get('strategy_tree',{}).get('branches'):
for b in d['strategy_tree']['branches']:
if b['id'] == branch_id:
b.setdefault('trigger_count', 0)
b['trigger_count'] += 1
b['last_trigger_price'] = round(price, 2)
b['last_triggered'] = datetime.now().isoformat()
break
json.dump(raw, open(DECISIONS_PATH, 'w'), ensure_ascii=False, indent=2)
except Exception:
pass
# ── 区间偏离检测 ──────────────────────────────────────────────────────────
def load_state():
@@ -254,9 +317,22 @@ def get_trigger_zones(trigger):
def run_once(round_label=""):
"""执行一轮完整的监控流程"""
global _SCENARIO_CACHE, _BRANCH_CACHE
label = f" [{round_label}]" if round_label else ""
start = time.time()
# 刷新情景与分支缓存（每轮更新）
_SCENARIO_CACHE = detect_scenario() if HAS_TREE else {}
_BRANCH_CACHE = {}
try:
raw = json.load(open(DECISIONS_PATH))
for d in raw.get('decisions', []):
tree = d.get('strategy_tree', {})
if tree and tree.get('branches'):
_BRANCH_CACHE[d['code']] = tree['branches']
except Exception:
pass
# === 第一步：一次性刷新所有价格 ===
refreshed = refresh_data_prices()
@@ -310,7 +386,8 @@ def run_once(round_label=""):
if in_zone and prev_in_zone != True:
if key == "stop_loss":
outputs.append(f"⚠️ {name}({code}) {price} → 跌破止损{hi}！")
branch_sfx = _branch_alert_suffix(code, price, d.get('shares',0), d.get('cost',0))
outputs.append(f"⚠️ {name}({code}) {price} → 跌破止损{hi}！{branch_sfx}")
record_event(code, name, "stop_loss", price, str(hi))
else:
extra = ""
@@ -323,7 +400,8 @@ def run_once(round_label=""):
act = trig.get("take_profit_action", "")
if act:
extra = f"（{act}）"
outputs.append(f"⚡ {name}({code}) {price} → 进入{label}{lo}~{hi}{extra}")
branch_sfx = _branch_alert_suffix(code, price, d.get('shares',0), d.get('cost',0))
outputs.append(f"⚡ {name}({code}) {price} → 进入{label}{lo}~{hi}{extra}{branch_sfx}")
record_event(code, name, "entry_zone", price, f"{lo}~{hi}", label)
state[code][key] = True
state_updated = True
@@ -410,25 +488,46 @@ def run_once(round_label=""):
except Exception as e:
outputs.append(f" ⚠️ 全量重评失败: {e}")
# === 第四步：输出 ===
# === 第四步：情景变化检测 + 输出 → 直接推XMPP ===
now_str = datetime.now().strftime("%H:%M:%S")
elapsed = time.time() - start
# 情景变化检测（跨轮对比）
if HAS_TREE and _SCENARIO_CACHE.get('id'):
prev_scenario = state.get('_system', {}).get('last_scenario', '')
curr_scenario = _SCENARIO_CACHE['id']
if prev_scenario and curr_scenario != prev_scenario:
combo = _SCENARIO_CACHE.get('combo_action', '')
outputs.insert(0, f"🌀 情景切换: {prev_scenario}→{curr_scenario} | {combo}")
if outputs:
state.setdefault('_system', {})['last_scenario'] = curr_scenario
state_updated = True
elif not prev_scenario:
state.setdefault('_system', {})['last_scenario'] = curr_scenario
state_updated = True
if outputs:
print(f"\n🔔 {now_str}{label}")
# 简短一行一个触发
for o in outputs:
print(o)
print(f"\n<structured_data>{json.dumps({'type':'价格监控','time':now_str,'triggers':outputs}, ensure_ascii=False)}</structured_data>")
else:
# 无触发时 SILENT（中继不推送）
print(f"[SILENT]{label} 价格正常 | {refreshed}只已刷新 | {elapsed:.1f}s")
# 直接推到老爸私信（免等待 cron_to_xmpp）
try:
body = "\n".join([f"{now_str}"] + outputs)
payload = json.dumps({
"to": "hmo@yoin.fun", "body": body, "type": "chat",
}).encode("utf-8")
req = urllib.request.Request(
"http://127.0.0.1:5805/", data=payload,
headers={"Content-Type": "application/json"},
)
urllib.request.urlopen(req, timeout=5)
except Exception:
pass
# else: SILENT — 无触发，无输出，不推
if state_updated:
save_state(state)
# 输出耗时
print(f"⏱{label} {elapsed:.1f}s", flush=True)
def main():
"""每cron触发跑一轮"""
strategy_lifecycle.py
+226 -5
View File
@@ -541,10 +541,29 @@ def reassess_strategy(code, name, price, cost, shares, current_action,
new_stop = trailing_stop
print(f" 已启用移动止损: {new_stop} (保护+{profit_pct:.1f}%利润)", file=sys.stderr)
# 最小止损距离：止损不能高于现价的97%（即至少3%下行空间）
min_stop = round(price * 0.97, 2)
# 最小止损距离 —— 随趋势强度调整（2026-06-23 震度保护规则）
# 强趋势（多周期看多 + MA多头排列）：最小1.5%下行空间
# 普通/弱势：最小3%下行空间
is_strong_trend = False
trend_align = mtf_adj.get("trend_alignment", "")
strong_trend_indicators = ["多周期看多", "多周期多头", "上升"]
try:
if any(ind in trend_align for ind in strong_trend_indicators) and ma20 > ma60 and cur >= ma20:
is_strong_trend = True
except (NameError, TypeError):
pass # ma20/ma60/cur may be unbound if MTF data insufficient
if is_strong_trend:
min_stop_gap = 0.015 # 1.5%
else:
min_stop_gap = 0.03 # 3%
min_stop = round(price * (1 - min_stop_gap), 2)
if new_stop > min_stop and not is_deep_loss:
old_stop = new_stop
new_stop = min_stop
if old_stop != new_stop:
print(f" 最小止损 {round(min_stop_gap*100)}%间距约束: {old_stop}→{new_stop} (趋势{'强' if is_strong_trend else '普通'})")
# ----- 止盈设置 -----
if is_short_term_strong_trend and not is_new_entry:
@@ -866,11 +885,155 @@ def load_fundamentals(code):
return {}
def _get_portfolio_risk_state():
"""读取 portfolio 组合风险状态（2026-06-23 引擎协调）"""
try:
# 数据一致性检查：警告多副本（2026-06-23 bugfix）
_check_portfolio_consistency()
p = json.load(open('/home/hmo/web-dashboard/data/portfolio.json'))
pos_pct = p.get('position_pct', 0)
cash = p.get('cash', 0)
holdings = p.get('holdings', [])
weak_cnt = sum(1 for h in holdings if h.get('change_pct', 0) < -15)
total = len(holdings) or 1
weak_ratio = weak_cnt / total
return {
'position_pct': pos_pct,
'cash': cash,
'is_high_position': pos_pct > 80,
'is_very_high_position': pos_pct > 90,
'is_high_weak': weak_ratio > 0.35,
'weak_ratio': round(weak_ratio * 100),
'total_holdings': total,
}
except:
return {}
def _is_buy_signal(signal):
"""判断信号是否为买入/持有类（用于防洗盘）"""
if not signal:
return False
buy_keywords = ['买入', '持有', '加仓', '阳线企稳', '温和放量', '量价齐升',
'缩量回踩', '接近支撑', '关注', '回调机会', '价稳']
for kw in buy_keywords:
if kw in signal:
return True
return False
def _check_portfolio_consistency():
"""数据一致性检查：如果存在多份 portfolio.json 则报警（2026-06-23 bugfix）"""
main = '/home/hmo/web-dashboard/data/portfolio.json'
main_cash = None
try:
import json
main_cash = json.load(open(main)).get('cash')
except Exception:
return
for path in [
'/home/hmo/data/portfolio.json',
'/home/hmo/projects/MoFin/data/portfolio.json',
'/home/hmo/web-dashboard.bak/data/portfolio.json',
]:
if os.path.exists(path):
try:
other = json.load(open(path))
if other.get('cash') != main_cash:
print(f"⚠️ 数据一致性: {os.path.realpath(path)} cash={other.get('cash')} ≠ 主文件 cash={main_cash} (需清理)", file=sys.stderr)
except Exception:
pass
def _check_contradiction(code, today_only=True):
"""反馈循环核——检查本股是否有刚卖出的记录
返回 dict or None:
- sold_reason: 'portfolio_trim'|'stop_loss'
- sold_at: 卖出日期
- days_ago: 卖出距今交易日数
- is_today: 是否今日卖出
- tag: 追加到信号的标注
"""
try:
from datetime import datetime, date
dec = json.load(open('/home/hmo/web-dashboard/data/decisions.json'))
for e in dec.get('decisions', []):
if e.get('code') != code:
continue
sold_at = e.get('sold_at', '')
if not sold_at:
return None
try:
sd = datetime.strptime(sold_at, '%Y-%m-%d').date()
td = date.today()
days = (td - sd).days
except:
return None
reason = e.get('sold_reason', 'portfolio_trim')
if reason == 'stop_loss':
tag = '止损离场（逻辑破坏，短期不关注）'
else:
tag = '组合减仓后关注（已清仓，等回踩确认）'
return {
'sold_reason': reason,
'sold_at': sold_at,
'days_ago': days,
'is_today': days == 0,
'tag': tag,
}
except:
return None
return None
def _get_sell_priority_list():
"""减仓优先级排序：深套>亏损>微盈>盈利（2026-06-23 反馈循环）
返回 [(code, name, change_pct, position_pct, priority_label), ...]
按卖出的优先顺序排列（最先应该卖的在最前）
"""
try:
p = json.load(open('/home/hmo/web-dashboard/data/portfolio.json'))
holdings = p.get('holdings', [])
ranked = []
for h in holdings:
chg = h.get('change_pct', 0)
pos = h.get('position_pct', 0)
if chg < -30:
label = '深套(>30%)，优先减'
rank = 0
elif chg < -20:
label = '深套(>20%)，优先减'
rank = 1
elif chg < -10:
label = '亏损，建议减'
rank = 2
elif chg < 0:
label = '微亏，可减'
rank = 3
elif chg < 10:
label = '微盈，持有'
rank = 4
else:
label = '盈利，最后减'
rank = 5
ranked.append((rank, h['code'], h.get('name',''), chg, pos, label))
ranked.sort(key=lambda x: (x[0], -x[4])) # 优先 rank, 其次仓位大优先
return [{'code':c,'name':n,'change_pct':chg,'position_pct':pos,'label':l}
for r,c,n,chg,pos,l in ranked]
except:
return []
def enrich_timing_signal(base_signal, macro_desc="", sector_note="",
profit_pct=0, stock_category="", is_new_entry=False,
fundamentals=None, news_sentiment=None,
timing_signal_override=None):
"""多因子合成timing_signal——大盘+行业+基本面+技术
timing_signal_override=None,
portfolio_context=None): # 新增参数
"""多因子合成timing_signal——大盘+行业+基本面+技术+组合风险
返回 (enriched_signal, factors_list)
- enriched_signal: 可读的多因子信号描述
@@ -960,6 +1123,21 @@ def enrich_timing_signal(base_signal, macro_desc="", sector_note="",
if base_signal and base_signal != "neutral":
factors.append(base_signal)
# 5.5 组合风险因子（2026-06-23 双引擎协调）
if portfolio_context and not is_new_entry:
if portfolio_context.get('is_very_high_position'):
factors.append("组合仓位极重(>90%)")
elif portfolio_context.get('is_high_position'):
factors.append("组合仓位偏重(>80%)")
if portfolio_context.get('is_high_weak'):
factors.append(f"弱势占{portfolio_context.get('weak_ratio')}%")
elif portfolio_context and is_new_entry:
# 新买入推荐：注明组合上下文
if portfolio_context.get('is_high_position'):
factors.append(f"仓{portfolio_context.get('position_pct')}%现金有限")
elif portfolio_context.get('is_high_weak'):
factors.append("组合风险信号")
# 如果没有足够因素，返回信号不充分
if not factors:
return "信号不充分", []
@@ -995,7 +1173,7 @@ def reassess_with_context(code, name, price, cost, shares, current_action,
news_sentiment = {}
fund = {}
enriched, _ = enrich_timing_signal(
enriched, factors = enrich_timing_signal(
base_signal=result.get("timing_signal", ""),
macro_desc=macro_desc,
sector_note=sector_note,
@@ -1004,9 +1182,52 @@ def reassess_with_context(code, name, price, cost, shares, current_action,
is_new_entry=is_watchlist,
fundamentals=fund,
news_sentiment=news_sentiment,
portfolio_context=_get_portfolio_risk_state(),
)
result["timing_signal"] = enriched
# 6. 防洗盘：信号不要一天一翻（2026-06-23）
# 如果旧信号是买入/持有类，新信号是谨慎/等待类，但中期趋势未破→维持旧信号
try:
dec = json.load(open('/home/hmo/web-dashboard/data/decisions.json'))
for e in dec.get('decisions', []):
if e.get('code') == code:
old_signal = e.get('timing_signal', '')
if old_signal and _is_buy_signal(old_signal) and not _is_buy_signal(enriched):
# 中等趋势检查：MA5 > MA20 + 多周期看多
mtf = result.get('multi_tf_context', '')
if '看多' in mtf or '多头' in mtf:
try:
closes = [float(k.split()[2]) for k in mtf.split('|') if 'MA5' in k]
except:
closes = []
has_uptrend = 'MA5' in mtf and 'MA20' in mtf
if has_uptrend:
print(f" 防洗盘: {old_signal}→保持旧信号（中期趋势完整）")
result["timing_signal"] = f"{old_signal}（正常回调价稳）"
break
except Exception as e:
print(f" 防洗盘跳过: {e}")
# 7. 反馈循环核：检查本股是否有刚卖出的记录（2026-06-23）
contradiction = _check_contradiction(code)
if contradiction and contradiction.get('is_today'):
# 今日刚卖出 → 不屏蔽信号，但必须自标注矛盾
print(f" 反馈循环: {contradiction.get('tag')} (sold_at={contradiction.get('sold_at')})")
if _is_buy_signal(result.get('timing_signal', '')):
result['action_note'] = contradiction['tag']
# 在 timing_signal 中追加反馈标注，供报告层可见
curr_signal = result.get('timing_signal', '')
if '⚠️' not in curr_signal:
result['timing_signal'] = f"⚠️{contradiction['tag']}|{curr_signal}"
elif contradiction:
# 非今日卖出但近期卖出 → 标注已清仓
print(f" 近期清仓: sold_at={contradiction.get('sold_at')} ({contradiction.get('days_ago')}日前)")
if _is_buy_signal(result.get('timing_signal', '')):
curr_signal = result.get('timing_signal', '')
if '已清仓' not in curr_signal:
result['timing_signal'] = f"已清仓，{curr_signal}"
# 重建 action 文本（同步多因子信号）
try:
if new_action_needs_refresh(result, {"source": "auto"}, price):
strategy_tree.py
+400
View File
@@ -0,0 +1,400 @@
#!/usr/bin/env python3
"""
strategy_tree.py — 情景化多分支策略决策引擎
核心理念：
每只股票不再只有一个买入区+止损，而是有一棵决策树。
每个分支 = {条件, 动作, 优先级, 触发统计}
当前宏观情景决定走哪个分支。
自成长：
→ 每次分支被触发，记录 trigger_count + 后续5日盈亏
→ success_rate < 30% 且触发≥5次 → 自动标记 pruning_candidate
→ 每周 pruning 时剪掉低效分支
数据存在 decisions.json 的 strategy_tree 字段。
"""
import json, os, sys, re
from datetime import datetime, date, timedelta
DECISIONS_PATH = "/home/hmo/web-dashboard/data/decisions.json"
PORTFOLIO_PATH = "/home/hmo/web-dashboard/data/portfolio.json"
MACRO_PATH = "/home/hmo/web-dashboard/data/macro_context.json"
MARKET_PATH = "/home/hmo/web-dashboard/data/market.json"
TREND_PATH = "/home/hmo/web-dashboard/data/trend_signals.json"
# ── 情景定义 ──────────────────────────────────────────────────────────────
SCENARIOS = [
{
"id": "sharp_decline",
"label": "急跌防御",
"desc": "大盘放量下跌，多板块共振杀跌",
"rules": {"mood": "bearish", "sector_crash": True},
"portfolio_action": "减仓至80%以下，优先出弱势深套",
},
{
"id": "weak_consolidation",
"label": "弱势震荡",
"desc": "大盘缩量阴跌，结构分化",
"rules": {"mood": "neutral", "breadth": "weak"},
"portfolio_action": "保持仓位90%以内，调结构",
},
{
"id": "sector_rotation",
"label": "板块轮动",
"desc": "大盘窄幅，强势板块切换",
"rules": {"mood": "neutral", "rotation": True},
"portfolio_action": "跟随板块切换，减旧加新",
},
{
"id": "bullish_recovery",
"label": "反弹上行",
"desc": "大盘放量上涨，情绪回暖",
"rules": {"mood": "bullish"},
"portfolio_action": "加仓至95%，追随趋势",
},
]
# ── 情景判定 ──────────────────────────────────────────────────────────────
def detect_scenario():
"""从宏观+市场数据判断当前情景
返回:
{"id": str, "label": str, "confidence": float, "portfolio_action": str}
"""
scenario_id = "weak_consolidation" # 默认
confidence = 0.5
try:
macro = json.load(open(MACRO_PATH))
market = json.load(open(MARKET_PATH))
except Exception:
return {"id": scenario_id, "label": "默认-弱势震荡", "confidence": 0.3, "portfolio_action": "观望"}
mood = market.get("mood", "").lower()
# Try to get trend from macro_context
structure = macro.get("structure", {})
overall = structure.get("overall", "").lower()
trend_desc = structure.get("description", "").lower()
# Check for sharp decline
if "bearish" in mood or "bearish" in overall:
if "crash" in trend_desc or "跌幅" in trend_desc or "恐慌" in trend_desc:
scenario_id = "sharp_decline"
confidence = 0.7
elif "弱势" in trend_desc or "疲弱" in trend_desc:
scenario_id = "weak_consolidation"
confidence = 0.6
else:
scenario_id = "weak_consolidation"
confidence = 0.5
elif "bullish" in mood or "bullish" in overall:
scenario_id = "bullish_recovery"
confidence = 0.6
elif "neutral" in mood:
# Check for rotation signals
try:
trend = json.load(open(TREND_PATH))
if trend.get("rotation_detected"):
scenario_id = "sector_rotation"
confidence = 0.5
except Exception:
pass
scenario_id = "weak_consolidation"
confidence = 0.4
sc = next((s for s in SCENARIOS if s["id"] == scenario_id), SCENARIOS[0])
return {
"id": scenario_id,
"label": sc["label"],
"desc": sc["desc"],
"confidence": round(confidence, 2),
"portfolio_action": sc["portfolio_action"],
}
# ── 分支评估 ──────────────────────────────────────────────────────────────
def evaluate_branches(code, scenario_id, price, shares, cost):
"""评估某只股票在当前情景下的所有分支
从 decisions.json 读取 strategy_tree.branches[]
返回: [{branch_id, action_type, action_detail, priority, applicable}]
"""
try:
dec = json.load(open(DECISIONS_PATH))
except Exception:
return []
entry = None
for e in dec.get("decisions", []):
if e.get("code") == code:
entry = e
break
if not entry:
return []
branches = entry.get("strategy_tree", {}).get("branches", [])
if not branches:
return []
results = []
for br in sorted(branches, key=lambda b: b.get("priority", 999)):
applicable = _check_branch_condition(br, scenario_id, price, shares, cost)
results.append({
"branch_id": br.get("id"),
"action_type": br.get("action", {}).get("type", "hold"),
"action_detail": br.get("action", {}),
"priority": br.get("priority", 999),
"rationale": br.get("rationale", ""),
"applicable": applicable,
})
return results
def _check_branch_condition(branch, scenario_id, price, shares, cost):
"""检查分支条件是否满足"""
cond = branch.get("condition", {})
required_scenario = cond.get("scenario", "")
if required_scenario and required_scenario != scenario_id:
return False
# Price conditions
price_cond = cond.get("price", "")
if price_cond:
ops = re.findall(r'([<>=!]+)\s*([\d.]+)', price_cond)
for op, val_str in ops:
val = float(val_str)
op = op.strip()
if op == "<" and not (price < val):
return False
if op == ">" and not (price > val):
return False
if op == "<=" and not (price <= val):
return False
if op == ">=" and not (price >= val):
return False
if op == "==" and not (abs(price - val) < 0.01):
return False
# Trend condition
trend = cond.get("trend", "")
if trend and trend == "uptrend":
pass # TODO: check multi_timeframe
# Loss condition
loss_pct = cond.get("loss_pct", "")
if loss_pct and cost > 0:
actual_loss = (price - cost) / cost * 100
if "<" in str(loss_pct):
limit = float(str(loss_pct).replace("<", "").replace("%", ""))
if not (actual_loss < limit):
return False
return True
# ── 分支触发记录 ──────────────────────────────────────────────────────────
def record_branch_trigger(code, branch_id):
"""记录分支被触发了一次，用于自成长统计"""
try:
dec = json.load(open(DECISIONS_PATH))
for e in dec.get("decisions", []):
if e.get("code") == code:
st = e.setdefault("strategy_tree", {})
for br in st.get("branches", []):
if br.get("id") == branch_id:
br["trigger_count"] = br.get("trigger_count", 0) + 1
br["last_triggered"] = datetime.now().isoformat()
break
break
json.dump(dec, open(DECISIONS_PATH, "w"), ensure_ascii=False, indent=2)
except Exception:
pass
# ── 分支剪枝（自成长核心）─────────────────────────────────────────────────
def prune_low_performance_branches(min_triggers=5, min_success_rate=0.3):
"""剪掉低成功率分支——自成长机制
条件：触发≥min_triggers 次 且 success_rate < min_success_rate
被剪的分支移入 history 字段，不打删除（可追溯）
"""
try:
dec = json.load(open(DECISIONS_PATH))
except Exception:
return []
pruned = []
for e in dec.get("decisions", []):
st = e.setdefault("strategy_tree", {})
branches = st.get("branches", [])
kept = []
for br in branches:
tc = br.get("trigger_count", 0)
sr = br.get("success_rate")
if sr is not None and tc >= min_triggers and sr < min_success_rate:
# 移入 history
history = st.setdefault("pruned_branches", [])
br["pruned_at"] = datetime.now().isoformat()
br["prune_reason"] = f"低成功率: {sr:.0%} (触发{tc}次)"
history.append(br)
pruned.append(f'{e.get("code")}:{br.get("id")} ({sr:.0%} < {min_success_rate:.0%})')
else:
kept.append(br)
st["branches"] = kept
if pruned:
json.dump(dec, open(DECISIONS_PATH, "w"), ensure_ascii=False, indent=2)
return pruned
# ── 初始化策略树（为一只票创建默认分支）─────────────────────────────────────
def init_default_branches(code, name, entry_low, entry_high, stop_loss, take_profit):
"""为 stock 创建默认多分支策略——由 per_stock_reassess 调用"""
base_price = (entry_low + entry_high) / 2 if entry_low and entry_high else 0
branches = []
# 分支0：止损（始终有效）
if stop_loss:
branches.append({
"id": f"{code}_stop_loss",
"condition": {"price": f"<{stop_loss}"},
"action": {"type": "sell", "amount": "all", "reason": "止损"},
"priority": 0,
"rationale": "止损保护本金",
"trigger_count": 0,
"success_rate": None,
"last_triggered": None,
})
# 分支1：回调买入（弱势情景适用）
if entry_low:
branches.append({
"id": f"{code}_buy_dip",
"condition": {"scenario": "weak_consolidation", "price": f"<={entry_high}", "price_lower": f">={entry_low}"},
"action": {"type": "buy", "amount": "normal", "limit": entry_low, "reason": "回调支撑买入"},
"priority": 1,
"rationale": "价格回调到支撑区，弱势市场低吸",
"trigger_count": 0,
"success_rate": None,
"last_triggered": None,
})
# 分支2：突破追涨（强势情景适用）
if take_profit:
branches.append({
"id": f"{code}_breakout_chase",
"condition": {"scenario": "bullish_recovery", "price": f">={take_profit}"},
"action": {"type": "buy", "amount": "normal", "limit": "market", "reason": "突破确认追涨"},
"priority": 2,
"rationale": "价格突破阻力，确认上升趋势后买入",
"trigger_count": 0,
"success_rate": None,
"last_triggered": None,
})
# 分支3：减仓（急跌情景适用）
branches.append({
"id": f"{code}_trim",
"condition": {"scenario": "sharp_decline", "loss_pct": "<-15%"},
"action": {"type": "sell", "amount": "half", "reason": "急跌降风险"},
"priority": 3,
"rationale": "急跌市场，深套股减半仓减少敞口",
"trigger_count": 0,
"success_rate": None,
"last_triggered": None,
})
# 分支4：止盈（浮盈较大）
if take_profit and entry_low:
branches.append({
"id": f"{code}_take_profit",
"condition": {"price": f">={take_profit}"},
"action": {"type": "sell", "amount": "half", "reason": "止盈锁利"},
"priority": 4,
"rationale": "达到目标价，减半仓锁定利润",
"trigger_count": 0,
"success_rate": None,
"last_triggered": None,
})
# 分支5：持有（默认）
branches.append({
"id": f"{code}_hold",
"condition": {},
"action": {"type": "hold", "reason": "无明确信号，继续持有"},
"priority": 99,
"rationale": "没有分支匹配时的默认动作",
"trigger_count": 0,
"success_rate": None,
"last_triggered": None,
})
return branches
# ── 组合约束检查 ──────────────────────────────────────────────────────────
def check_portfolio_constraint(action_type, amount, cash_remain=None):
"""组合约束检查：现金够不够？仓位上限？"""
try:
pf = json.load(open(PORTFOLIO_PATH))
except Exception:
return True, "无法读取组合"
if action_type == "buy":
# 估算买入金额
cost_est = amount if amount else 100000 # default 10万
if cash_remain is not None:
cost_est = cash_remain
if cost_est > pf.get("cash", 0):
return False, f"现金不足: 需要~{cost_est:.0f}，可用{pf['cash']:.0f}"
return True, "OK"
# ── CLI 入口 ──────────────────────────────────────────────────────────────
if __name__ == "__main__":
import argparse
parser = argparse.ArgumentParser(description="多分支策略决策引擎")
parser.add_argument("--detect", action="store_true", help="检测当前情景")
parser.add_argument("--evaluate", type=str, help="评估指定股票的分支")
parser.add_argument("--prune", action="store_true", help="剪枝低效分支")
args = parser.parse_args()
if args.detect:
sc = detect_scenario()
print(f"情景: {sc['id']} ({sc['label']})")
print(f"置信度: {sc['confidence']}")
print(f"组合动作: {sc['portfolio_action']}")
if args.evaluate:
code = args.evaluate
sc = detect_scenario()
print(f"当前情景: {sc['id']} ({sc['label']})")
print(f"评估 {code}:")
results = evaluate_branches(code, sc["id"], 0, 0, 0)
for r in results:
status = "✅" if r["applicable"] else " "
print(f" {status} [{r['priority']}] {r['branch_id']} → {r['action_type']}: {r['rationale']}")
if args.prune:
pruned = prune_low_performance_branches()
if pruned:
print(f"已剪枝: {len(pruned)} 条")
for p in pruned:
print(f" - {p}")
else:
print("无需要剪枝的分支")