chore: 重新归档meta_growth/meta_watchdog/ab_research_daily(上次被deploy_guard回滚)

This commit is contained in:
xxm
2026-08-21 11:08:00 +08:00
parent 15edc92371
commit b3cfe59770
5 changed files with 1630 additions and 0 deletions
File diff suppressed because it is too large Load Diff
@@ -0,0 +1,95 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""AB路线每日LLM主导研究 v2(老莫2026-08-18
在原规则化分析基础上,集成 LLM 生成深度研究结论(真正"LLM主导"
1. 读温区覆盖 + 进化中心 + B组候选
2. LLM 分析薄弱环节 → 建议尝试
3. 写 strategy_research_log 表
"""
import sys, os, json, sqlite3
from datetime import datetime
sys.path.insert(0, "/home/hmo/MoFin/deploy/profile-scripts")
DB = "/home/hmo/MoFin/data/mofin.db"
CENTER = "/home/hmo/MoFin/data/evolution_center.json"
def ensure_table(conn):
conn.execute("""CREATE TABLE IF NOT EXISTS strategy_research_log (
id INTEGER PRIMARY KEY AUTOINCREMENT, log_date TEXT NOT NULL, market TEXT,
weak_regime TEXT, finding TEXT, experiment TEXT, result TEXT,
produced_strategy TEXT, produced_verified INTEGER DEFAULT 0, llm_model TEXT, created_at TEXT)""")
conn.commit()
def load_center():
if not os.path.exists(CENTER): return {}
try: return json.load(open(CENTER))
except: return {}
def build_prompt(coverage, center):
"""构造 LLM 研究 prompt"""
line = []
line.append("你是MoFin策略研究员。分析当前策略覆盖,找出薄弱环节并给出研究建议。")
line.append("温区覆盖(trades>=302y):")
for c in coverage:
line.append(f"- {c['market']}/{c['regime']}: {c['count']}个策略")
bg = center.get("b_group") or []
if bg:
line.append(f"B组候选: {len(bg)}")
for b in bg[:3]:
line.append(f" - {str(b)[:80]}")
line.append("\n请输出:")
line.append("1. 最薄弱的温区/环节(策略匮乏或合格策略少)")
line.append("2. 具体研究建议(做什么尝试)")
line.append("3. 预期成果类型")
line.append("格式:发现|建议|预期")
return "\n".join(line)
def analyze_llm(coverage, center):
"""LLM 生成研究结论"""
try:
from llm_client import call_llm
prompt = build_prompt(coverage, center)
res = call_llm(prompt)
return str(res)[:400] if res else None
except Exception as e:
return f"[LLM调用失败: {e}]"
def main():
conn = sqlite3.connect(DB, timeout=30)
conn.execute("PRAGMA busy_timeout=30000")
ensure_table(conn)
today = datetime.now().strftime("%Y-%m-%d")
if conn.execute("SELECT COUNT(*) FROM strategy_research_log WHERE log_date=?", (today,)).fetchone()[0]:
print(f"[AB研究] {today} 已有记录"); conn.close(); return
# 读取覆盖
rows = conn.execute("""SELECT market, regime, COUNT(DISTINCT strategy) as cnt
FROM strategy_regime_perf_by_period WHERE period_tag='2y' AND trades >= 30
GROUP BY market, regime""").fetchall()
coverage = [{"market": r[0], "regime": r[1], "count": r[2]} for r in rows]
center = load_center()
# 基础规则发现
findings = []
weak = []
if coverage:
c_sorted = sorted(coverage, key=lambda x: x["count"])
weak = c_sorted[:2]
findings.append("覆盖最少的温区: " + "; ".join(f"{c['market']}/{c['regime']}({c['count']})" for c in weak))
# LLM 深度分析
if coverage:
llm_res = analyze_llm(coverage, center)
if llm_res:
findings.append("LLM分析: " + llm_res)
finding_text = "; ".join(findings) or "无明显薄弱点"
weak_rg = weak[0]["regime"] if weak else ""
weak_mkt = weak[0]["market"] if weak else "a"
conn.execute(
"INSERT INTO strategy_research_log (log_date, market, weak_regime, finding, experiment, result, produced_strategy, created_at) "
"VALUES (?,?,?,?,?,?,?,?)",
(today, weak_mkt, weak_rg, finding_text, "LLM主导温区覆盖+B组分析", "记录待验证", "",
datetime.now().isoformat()))
conn.commit()
print(f"[AB研究] {today} 记录完成 (LLM主导)")
conn.close()
if __name__ == "__main__":
main()
+250
View File
@@ -0,0 +1,250 @@
#!/usr/bin/env python3
"""
meta_growth.py — 自成长机制的元层
功能:读取近期 git log,识别修复模式,注入新扫描规则到 hardcode_scanner 的扩展点。
让自成长机制本身也会成长——能自动发现新的问题类型并添加对应的扫描规则。
调度:交易日 12:45 和 00:45no_agent 模式)
- 12:45: 上午盘发现的问题→下午17:25审计就能扫到
- 00:45: 全天修复汇总→次日审计带新规则
输出:/home/hmo/web-dashboard/data/growth_registry.json
"""
import subprocess
import re
import json
import os
import sys
import datetime
SCANNER_PATH = "/home/hmo/MoFin/deploy/profile-scripts/hardcode_scanner.py"
PROFILE_SCANNER = "/home/hmo/.hermes/profiles/position-analyst/scripts/hardcode_scanner.py"
REGISTRY_PATH = "/home/hmo/web-dashboard/data/growth_registry.json"
EXTENSION_MARKER = "# 扩展点 — meta_growth 在此追加新规则"
# 已知问题类别 → 扫描规则模板
# meta_growth 分析 git log 后,把新模式匹配到这里生成规则元组
PATTERN_TEMPLATES = [
{
"name": "cash_hardcode",
"desc": "现金/金额硬编码",
"regex": r"return\s+\d{4,}\b",
"reason": "可能的硬编码现金/金额",
"git_keywords": ["cash", "现金", "硬编码", "金额", "fallback.*\\d+"],
},
{
"name": "exchange_rate",
"desc": "汇率硬编码",
"regex": r"0\.8[5-9]\d{1,3}",
"reason": "可能的硬编码汇率值",
"git_keywords": ["汇率", "rate", "HKD", "CNY", "0.8[5-9]"],
},
{
"name": "lot_size_hardcode",
"desc": "港股每手股数硬编码",
"regex": r"1手\s*[:=]\s*\d{3,}",
"reason": "可能的每手股数硬编码",
"git_keywords": ["lot_size", "每手", "手数", "lot", "board lot", "f\\[60\\]"],
},
{
"name": "percent_threshold",
"desc": "百分比阈值硬编码",
"regex": r"[><=]\s*0\.[0-9]+",
"reason": "可能的百分比阈值硬编码",
"git_keywords": ["threshold", "阈值", "止损", "stop_loss", "止盈", "百分比"],
},
{
"name": "position_limit",
"desc": "仓位金额硬编码",
"regex": r"仓位\s*[:=]\s*\d{3,}",
"reason": "可能的仓位金额硬编码",
"git_keywords": ["仓位", "position", "持仓金额"],
},
{
"name": "hardcoded_path",
"desc": "路径硬编码",
"regex": r"['\"](?!http|~|\./|\.\./)/home/[^'\"]+['\"]",
"reason": "可能的文件路径硬编码(应使用环境变量或配置)",
"git_keywords": ["路径", "path", "hardcoded path"],
},
]
def get_recent_git_log(hours=8):
"""获取最近 N 小时的 git log"""
try:
result = subprocess.run(
["git", "log", f"--since={hours} hours ago", "--oneline"],
capture_output=True, text=True, cwd="/home/hmo/MoFin", timeout=10
)
return result.stdout
except Exception as e:
print(f"[meta_growth] git log 失败: {e}", file=sys.stderr)
return ""
def analyze_log(log_text):
"""分析 git log,识别修复模式"""
found_patterns = []
lines = log_text.strip().split("\n")
for tmpl in PATTERN_TEMPLATES:
hit_count = 0
for line in lines:
for kw in tmpl["git_keywords"]:
if re.search(kw, line, re.IGNORECASE):
hit_count += 1
break
if hit_count > 0:
found_patterns.append({
"name": tmpl["name"],
"desc": tmpl["desc"],
"regex": tmpl["regex"],
"reason": tmpl["reason"],
"hits": hit_count,
})
return found_patterns
def load_registry():
"""加载问题类别注册表"""
try:
if os.path.exists(REGISTRY_PATH):
with open(REGISTRY_PATH) as f:
return json.load(f)
except Exception:
pass
return {
"known_categories": [],
"injected_rules": [],
"last_run": None,
"last_findings": [],
}
def save_registry(registry):
"""保存注册表"""
os.makedirs(os.path.dirname(REGISTRY_PATH), exist_ok=True)
with open(REGISTRY_PATH, "w") as f:
json.dump(registry, f, indent=2, ensure_ascii=False)
def rule_already_exists(registry, regex):
"""检查规则是否已注入"""
for r in registry.get("injected_rules", []):
if r.get("regex") == regex:
return True
return False
def inject_rule(scanner_path, regex, reason, marker=EXTENSION_MARKER):
"""在 hardcode_scanner.py 的扩展点后插入新规则"""
if not os.path.exists(scanner_path):
return False
try:
with open(scanner_path, "r") as f:
content = f.read()
except Exception:
return False
if regex in content:
return False # 已存在
new_rule = f" (r'{regex}', '{reason}'),\n {marker}"
if marker not in content:
return False # 没有扩展点
content = content.replace(marker, new_rule)
try:
with open(scanner_path, "w") as f:
f.write(content)
return True
except Exception:
return False
def self_check():
"""自检:检查自成长系统本身的健康度"""
issues = []
if not os.path.exists(SCANNER_PATH):
issues.append("hardcode_scanner.py 不存在")
if not os.path.exists(REGISTRY_PATH):
issues.append("growth_registry.json 不存在(首次运行正常)")
return issues
def main():
now = datetime.datetime.now().isoformat()
period = "afternoon" if datetime.datetime.now().hour < 15 else "overnight"
# 自检
issues = self_check()
if issues:
for issue in issues:
print(f"[meta_growth] ⚠ {issue}", file=sys.stderr)
# 读取 git log
hours = 8 # 过去8小时(覆盖一整个交易时段)
log = get_recent_git_log(hours=hours)
if not log:
print(f"[meta_growth] 无近期提交,跳过")
return
print(f"[meta_growth] 分析 {period} 时段日志 ({len(log.strip().split(chr(10)))} 条提交)")
# 分析修复模式
patterns = analyze_log(log)
# 加载注册表
registry = load_registry()
registry["last_run"] = now
if not patterns:
print(f"[meta_growth] 未发现新修复模式")
registry["last_findings"] = []
save_registry(registry)
return
# 去重注入
injected_count = 0
for p in patterns:
if rule_already_exists(registry, p["regex"]):
print(f"[meta_growth] 规则已存在: {p['name']} ({p['regex']})")
continue
# 注入到 MoFin and profile 两个副本
ok1 = inject_rule(SCANNER_PATH, p["regex"], p["reason"])
ok2 = inject_rule(PROFILE_SCANNER, p["regex"], p["reason"])
if ok1 or ok2:
registry["injected_rules"].append({
"name": p["name"],
"desc": p["desc"],
"regex": p["regex"],
"reason": p["reason"],
"injected_at": now,
"period": period,
"hits_in_log": p["hits"],
})
injected_count += 1
print(f"[meta_growth] 注入新规则: {p['name']} ({p['desc']})")
# 记录到已知类别
if p["name"] not in registry["known_categories"]:
registry["known_categories"].append(p["name"])
registry["last_findings"] = patterns
save_registry(registry)
print(f"[meta_growth] 本次注入 {injected_count} 条新规则")
if injected_count > 0:
print(f"[meta_growth] 下次 hardcode_scanner 运行时将自动使用新规则")
if __name__ == "__main__":
main()
+121
View File
@@ -0,0 +1,121 @@
#!/usr/bin/env python3
"""meta_watchdog.py — L4 自检系统的自检(看门狗的看门狗)
检查 L1-L3 各自检组件本身是否在正常运转:
- L1 functional_health_check: functional_health.json 是否 <20min(交易时段)
- L2 system_hygiene_audit: hygiene_report.json 是否 <26h(每日)
- L3 self_repair: repair_state.json 存在性 + cron 是否注册
- mofin_health 采集: mofin_health.json 是否 <20min(交易时段)
- XMPP 桥: :5805 是否可发(self_repair 的报备通道)
2026-08-13 删除 L0 agents_health_check 检查项:MoFin 无该组件,且 L1 functional_health_check 已覆盖健康检查功能,检查项是死代码)
任何一层死了 → 推 XMPP 点名(这是最后的兜底,必须直达用户)。
频率:每小时(cron)。输出 gateway/logs/meta_watchdog.json。
"""
import os, sys, json, subprocess
from datetime import datetime
OUT = '/home/hmo/MoFin/gateway/logs/meta_watchdog.json'
LAYERS = [
{"layer": "L1 functional_health", "file": "/home/hmo/MoFin/gateway/logs/functional_health.json",
"max_age_min": 25, "when": "trading",
"repair": "L1 cron 停摆,检查 hermes cron 引擎"},
{"layer": "L2 hygiene_audit", "file": "/home/hmo/MoFin/gateway/logs/hygiene_report.json",
"max_age_min": 26 * 60, "when": "always",
"repair": "L2 每日审计未跑,检查 hermes cron"},
{"layer": "L1.5 mofin_health采集", "file": "/home/hmo/web-dashboard/static/mofin_health.json",
"max_age_min": 25, "when": "trading",
"repair": "mofin_health.py 采集停摆"},
{"layer": "L3 self_repair", "file": "/home/hmo/MoFin/gateway/logs/repair_log.jsonl",
"max_age_min": None, "when": "meta",
"repair": "self_repair cron 未注册"},
]
def is_trading(now):
return now.weekday() < 5 and 9 <= now.hour <= 16
def main():
now = datetime.now()
trading = is_trading(now)
results = []
for L in LAYERS:
if L["when"] == "trading" and not trading:
results.append({"layer": L["layer"], "status": "skip", "reason": "非交易时段"})
continue
if L["when"] == "meta":
# 检查 self_repair 是否注册在 cron
try:
d = json.load(open('/home/hmo/.hermes/profiles/position-analyst/cron/jobs.json'))
jobs = d if isinstance(d, list) else d.get('jobs', [])
registered = any(j.get('script') == 'self_repair.py' and j.get('enabled', True) for j in jobs)
results.append({"layer": L["layer"],
"status": "ok" if registered else "fail",
"reason": "已注册" if registered else "未在 cron 注册"})
except Exception as e:
results.append({"layer": L["layer"], "status": "fail", "reason": str(e)[:60]})
continue
f = L["file"]
if not os.path.exists(f):
results.append({"layer": L["layer"], "status": "fail",
"reason": f"输出物不存在", "repair": L["repair"]})
continue
age_min = (now.timestamp() - os.path.getmtime(f)) / 60
if L["max_age_min"] and age_min > L["max_age_min"]:
results.append({"layer": L["layer"], "status": "fail",
"reason": f"输出物 {age_min/60:.1f}h 未更新(阈值 {L['max_age_min']}min",
"repair": L["repair"]})
else:
results.append({"layer": L["layer"], "status": "ok",
"reason": f"{age_min:.0f}min 前"})
# XMPP 桥(报备通道):只收 POSTGET 会 501,但任何 HTTP 响应都说明进程活着
try:
import urllib.request
urllib.request.urlopen('http://127.0.0.1:5805/', timeout=3)
results.append({"layer": "XMPP桥 :5805", "status": "ok", "reason": "可达"})
except urllib.error.HTTPError as e:
results.append({"layer": "XMPP桥 :5805", "status": "ok", "reason": f"可达(HTTP {e.code})"})
except Exception:
results.append({"layer": "XMPP桥 :5805", "status": "fail",
"reason": "不可达", "repair": "重启 xmpp-zhiwei"})
fails = [r for r in results if r["status"] == "fail"]
report = {
"generated_at": now.strftime("%Y-%m-%d %H:%M:%S"),
"status": "fail" if fails else "ok",
"layers": results,
}
os.makedirs(os.path.dirname(OUT), exist_ok=True)
with open(OUT, 'w', encoding='utf-8') as f:
json.dump(report, f, ensure_ascii=False, indent=2)
print(f"meta_watchdog: {report['status']}")
for r in results:
icon = {"ok": "", "fail": "", "skip": ""}[r["status"]]
print(f" {icon} {r['layer']}: {r['reason']}")
if fails:
try:
import urllib.request
lines = [f"🚨 自检系统自检(L4兜底)发现 {len(fails)} 层异常:"]
for r in fails:
lines.append(f"{r['layer']}: {r['reason']}")
if r.get('repair'):
lines.append(f" → 处置建议: {r['repair']}")
payload = json.dumps({'to': 'hmo@yoin.fun', 'body': '\n'.join(lines), 'type': 'chat'}).encode()
req = urllib.request.Request('http://127.0.0.1:5805/', data=payload,
headers={'Content-Type': 'application/json'})
urllib.request.urlopen(req, timeout=5)
print(' 📨 已推 XMPP(兜底直达)')
except Exception as e:
print(f' XMPP 失败: {e}')
if __name__ == '__main__':
main()
+80
View File
@@ -0,0 +1,80 @@
# 代码结构大扫除(文件级清理)2026-08-20
> 目标:消除散落的多套副本、死代码、多余硬链接入口,收敛为「单点权威 + 运行副本」的干净结构。
## 一、清理前摸底(5 个散落位置)
| 目录 | 数量 | 性质 |
|---|---|---|
| `deploy/profile-scripts/` | 176 | **git 权威源码**(唯一修改点)|
| `.hermes/.../scripts/` | 292 | cron 运行目录 = 176 硬链接 + **116 独有** |
| `MoFin/` 根 | 40 | 19 硬链接 + 2 旧版 + 19 独有应用模块 |
| `MoFin/scripts/` | 3 | 硬链接 + prepare_report_data 重复副本 |
| `/home/hmo/scripts/` | 15 | 无引用的旧项目残留 |
## 二、本次清理动作(已物理生效,archive 可还原)
### 1. 归档 hermes 独有死工具 111 个 → `archive/hermes-dead-tools-20260820/`
- 特征:不在 cron 调度、不被任何活跃脚本 import 的一次性排查/测试/清理工具
`check_*`/`test_*`/`audit_*`/`fix_*`/`sc*`/`verify_*`/`remove_*` 等)
- 含 4 个废弃 scanner`btd1_v3_scanner`/`market_scanner`/`market_thermometer`cron note 已注明改为 market_regime 废弃)/`s2v2_scanner`
- **hermes 独有从 116 收敛到 5 个核心**:
`alert_logger`(被 price_monitor import)、`market_screener`/`prepare_report_data`/`self_todo_executor_v2`cron 调度)、`xmpp_zhiwei_bot`XMPP bot wrapper
### 2. 归档 MoFin 根 2 个旧版 → `archive/legacy-cleanup-20260820/mofin-root-old/`
- `mo_models.py``technical_analysis.py`deploy 版本更新(8-14)且为权威,根旧版可删
### 3. 归档 `/home/hmo/scripts/` 整个目录(无引用旧项目)→ `archive/legacy-cleanup-20260820/home-scripts/`
- crontab 引用 0、systemd 引用 0,确认无引用
### 4. 收敛 `MoFin/scripts/prepare_report_data.py` 重复副本 → archive
- cron 用 hermes 那份(jobs.py 从 hermes/scripts 解析),MoFin/scripts 是内容相同的重复,归档
## 三、关于硬链接的结论(沉淀知识)
**为什么需要这么多硬链接?—— 不是冗余,是 server.py 的多层 sys.path 设计所需。**
- `server.py` 的 sys.path 依次注入:`MoFin/scripts``MoFin/``MoFin/deploy/profile-scripts`
- `mofin_db.py`/`mo_data.py` 被 MoFin 根 20+ 个模块 + deploy 40+ 个脚本共同依赖
- **用硬链接(同 inode** 让每个 sys.path 目录都能找到同一份代码,**共享数据块不占额外空间**,改一处处处生效
- **结论:`mofin_db.py`/`mo_data.py` 的多硬链接入口是合理设计,保留**;真正冗余的是「内容相同但 inode 不同的独立副本」,已清理
## 四、收敛后的目标结构(代码结构规范)
```
deploy/profile-scripts/ ← 唯一 git 权威源码(修改只在这里做)
└─ *.py 176 个
.hermes/.../scripts/ ← cron 实际运行目录(176 硬链接 + 5 核心独有)
MoFin/ ← 应用主程序(server.py 等,独立于 cron 脚本集)
MoFin/scripts/ ← 仅部署脚本 + mofin_db/mo_data 硬链接
archive/cleansweep-2026*/ ← 归档的可还原死代码
```
## 五、铁律(防止再散落)
1. **新增/修改代码只进 `deploy/profile-scripts/`**git 权威)
2. **不在该目录下的运行时脚本一律视为孤儿**,先查 cron/import 引用再处理
3. **不新建内容重复的独立副本**;需要多目录访问同一代码时用**硬链接**(同 inode)
4. **一次性排查/测试工具用完即归档**,不留在运行目录
5. 归档统一放 `archive/`,标注日期,杜绝再次散落
## 六、验证(全部通过 ✅)
- deploy 权威 176 个完整;hermes 181 = 176 + 5 核心
- cron 全部脚本引用无缺失(jobs.json 每个 script 都存在)
- 关键模块 import 正常(price_monitor/mofin_health/anomaly_monitor/market_screener
- archive 可还原:111hermes 死工具)+ 30(收敛产物)
## 七、待办(受提交白名单限制)
- MoFin 仓库启用提交白名单(2026-07-21 批准),当前自主执行 agent 无 git 写权限
- **需 kanban 提单给莫笑笑评审,由她 commit 本清理**
- 归档目录已在磁盘物理持久化,cron 未受影响,可安全等待评审
## 八、复查补充(2026-08-20 收尾复查)
### 保留说明:`fix_gateway.py`(勿删)
- 虽然它不在 cron、名字像 `fix_*` 死工具,但**它是 `fix_gateway_port.py` 的活跃依赖**(被 import)。
- cron job「Gateway看门狗-知微」执行 `fix_gateway_port.py`enabled),内部 import `fix_gateway`
- **结论:`fix_gateway.py`hermes + deploy 各一份,与 fix_gateway_port 同源)必须保留**,不是死工具。
- 复查确认:无漏网死工具;deploy=176 / hermes=181176+5核心)结构正确。