diff --git a/scripts/branch_scanner.py b/scripts/branch_scanner.py new file mode 100644 index 0000000..755d700 --- /dev/null +++ b/scripts/branch_scanner.py @@ -0,0 +1,68 @@ +#!/usr/bin/env python3 +"""每30分钟扫描所有持仓的分支状态，发现可操作的分支就推""" +import sys, json, os +sys.path.insert(0, '/home/hmo/MoFin') +sys.path.insert(0, '/home/hmo/web-dashboard') +from strategy_tree import detect_scenario + +DEC_PATH = '/home/hmo/web-dashboard/data/decisions.json' +PF_PATH = '/home/hmo/web-dashboard/data/portfolio.json' +XMPP_BRIDGE = "http://127.0.0.1:5805/" +XMPP_USER = "hmo@yoin.fun" + +def push(msg): + try: + from urllib.request import Request, urlopen + payload = json.dumps({"to": XMPP_USER, "body": msg, "type": "chat"}).encode() + req = Request(XMPP_BRIDGE, data=payload, headers={"Content-Type": "application/json"}) + urlopen(req, timeout=5) + except Exception: + pass + +def main(): + scenario = detect_scenario() + if not scenario.get('id'): + return 0 # SILENT + + alives = [] + dec = json.load(open(DEC_PATH)) + pf = json.load(open(PF_PATH)) + pf_codes = {h['code'] for h in pf.get('holdings', [])} + + for e in dec.get('decisions', []): + code = e.get('code', '') + tree = e.get('strategy_tree', {}) + if not tree or not tree.get('branches'): + continue + branches = tree['branches'] + price = e.get('price', 0) + shares = e.get('shares', 0) + cost = e.get('cost', 0) + if price <= 0: + continue + # 用 strategy_tree 评估 + try: + from strategy_tree import evaluate_branches + results = evaluate_branches(code, scenario['id'], price, shares, cost) + for r in results: + if r.get('applicable') and r.get('action_type') != 'hold': + is_held = code in pf_codes + label = '持仓' if is_held else '自选' + alives.append(f" {label} {r.get('action_type','?')} {code}@{price} | 情景{scenario.get('label','')}→{r.get('branch_id','').split('_')[-1]}| {r.get('rationale','')[:30]}") + break + except Exception: + pass + + if not alives: + return 0 # SILENT + + # 只推单一情景行 + 操作列表 + out_lines = [f"【知微】分支扫描 | {scenario.get('label','')}({scenario.get('id','')})"] + out_lines.extend(alives) + msg = '\n'.join(out_lines) + print(msg) + push(msg) + return 1 + +if __name__ == '__main__': + sys.exit(main()) diff --git a/scripts/hardcode_scanner.py b/scripts/hardcode_scanner.py new file mode 100644 index 0000000..0c7fa2c --- /dev/null +++ b/scripts/hardcode_scanner.py @@ -0,0 +1,109 @@ +#!/usr/bin/env python3 +""" +hardcode_scanner.py — 自成长扫描器 + +检测脚本中可能已过时的硬编码数值，写审计 JSON 供 system_audit 调用。 + +扫描规则： +1. 财务类硬编码（cash/金额/仓位）— 应来自 data/*.json +2. 汇率类硬编码（0.86xx, 0.87xx, 0.93等）— 应来自 hk_rate 模块 +3. 数字 fallback（return X, fallback=X）— 应来自实时数据源 +4. 每手股数硬编码（500, 1000 等）— 应来自 Tencent API field[60] + +输出：/home/hmo/web-dashboard/data/hardcode_audit.json +""" +import re, ast, json, os, sys + +SCAN_DIRS = [ + "/home/hmo/.hermes/profiles/position-analyst/scripts", + "/home/hmo/MoFin", + "/home/hmo/web-dashboard", +] + +SAFE_FALLBACK_PATTERNS = [ + # Known valid fallbacks where network data is genuinely optional + "rate = 0.87", # hk_rate module's own fallback + "retry_for_secs=5", # timeouts + "timeout=5", + "timeout=10", + "timeout=30", + "timeout=60", + "port 5805", + "127.0.0.1:5805", +] + +SUSPICIOUS_NUMBERS = [ + # (pattern, reason) + (r'return\s+\d{4,}\b', '可能的硬编码现金/金额'), + (r'=\s*\d{5,}\b', '可能的硬编码大额数字'), + (r'0\.8[5-9]\d{1,3}', '可能的硬编码汇率值'), + (r'0\.9[0-5]\d{1,3}', '可能的硬编码汇率值'), + (r'1手\s*[:=]\s*\d{3,}', '可能的每手股数硬编码'), +] + +def scan_file(filepath): + findings = [] + try: + with open(filepath, 'r', encoding='utf-8', errors='ignore') as f: + content = f.read() + except Exception: + return [] + + lines = content.split('\n') + for i, line in enumerate(lines, 1): + stripped = line.strip() + # Skip comments and empty lines + if not stripped or stripped.startswith('#') or '"""' in stripped: + continue + + for pat, reason in SUSPICIOUS_NUMBERS: + if re.search(pat, stripped): + # Check if it's a safe fallback + if any(safe in stripped for safe in SAFE_FALLBACK_PATTERNS): + continue + findings.append({ + "file": filepath, + "line": i, + "code": stripped[:120], + "reason": reason, + "suggestion": "考虑从 data/*.json 或 API 实时读取，不使用硬编码值" + }) + break # one finding per line + return findings + + +def main(): + all_findings = [] + for directory in SCAN_DIRS: + if not os.path.isdir(directory): + continue + for root, _, files in os.walk(directory): + for f in files: + if f.endswith('.py'): + path = os.path.join(root, f) + findings = scan_file(path) + all_findings.extend(findings) + + # Only output to stdout for cron + if all_findings: + print(f"[HARDCODE_SCAN] 发现 {len(all_findings)} 处可能硬编码：") + for f in all_findings: + rel = f['file'].replace('/home/hmo/', '') + print(f" ⚠ {rel}:L{f['line']} {f['reason']}") + print(f" {f['code']}") + print(f" → {f['suggestion']}") + else: + print("[HARDCODE_SCAN] 未发现可疑硬编码") + + # Write audit log + os.makedirs(os.path.dirname(AUDIT_PATH), exist_ok=True) + json.dump({ + "timestamp": __import__('datetime').datetime.now().isoformat(), + "findings": all_findings, + "count": len(all_findings), + }, open(AUDIT_PATH, 'w'), ensure_ascii=False, indent=2) + + +if __name__ == '__main__': + AUDIT_PATH = "/home/hmo/web-dashboard/data/hardcode_audit.json" if 'AUDIT_PATH' not in dir() else AUDIT_PATH + main() diff --git a/scripts/prune_branches.py b/scripts/prune_branches.py new file mode 100755 index 0000000..e9c285c --- /dev/null +++ b/scripts/prune_branches.py @@ -0,0 +1,21 @@ +#!/usr/bin/env python3 +""" +prune_branches.py — 每周六凌晨6点执行 + +低效分支剪枝： +- 触发≥5次且成功率<30% → 剪枝归档 +- 输出报告（stdout → cron delivery） +""" +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) + +if result: + print(f"[分支剪枝] 已剪枝 {len(result)} 条低效分支：") + for item in result: + print(f" ✂ {item}") + print("[分支剪枝] 剪枝完成") +else: + print("[分支剪枝] 无低效分支需剪枝")