#!/usr/bin/env python3 """ per_stock_reassess.py — 按个股触发重评 对每只传进来的 code 执行 reassess_with_context(),然后写入 DB holding_strategies 表(纯DB模式,已移除JSON依赖)。 """ import sys, json, os, re sys.path.insert(0, "/home/hmo/web-dashboard") sys.path.insert(0, "/home/hmo/MoFin") from strategy_lifecycle import reassess_with_context as reassess_strategy from mo_data import read_decisions, read_portfolio def _build_full_analysis(code, entry, result): """从重评结果构建完整九维分析文本""" if not result: return "" lines = [] name = entry.get("name", code) price = result.get("price") or entry.get("price", 0) tech = result.get("tech_snapshot") or entry.get("tech_snapshot", "") sector = result.get("sector_context") or entry.get("sector_context", "") signal = result.get("timing_signal") or entry.get("timing_signal", "") category = result.get("stock_category") or entry.get("stock_category", "") el = result.get("entry_low") or entry.get("entry_low", 0) eh = result.get("entry_high") or entry.get("entry_high", 0) sl = result.get("stop_loss") or entry.get("stop_loss", 0) tp = result.get("take_profit") or entry.get("take_profit", 0) rr = result.get("rr_ratio") or entry.get("rr_ratio", 0) act = result.get("action", "") # 从tech_snapshot提取关键数据 ma5 = ma10 = ma20 = ma60 = "?" import re ma_match = re.search(r'MA5=([\d.]+).*?MA10=([\d.]+).*?MA20=([\d.]+).*?MA60=([\d.]+)', tech) if ma_match: ma5, ma10, ma20, ma60 = ma_match.groups() lines.append(f"【{name}({code}) 九维全析】") lines.append("") lines.append(f"① 大盘×技术面:价格{price},MA5={ma5} MA10={ma10} MA20={ma20} MA60={ma60}") if el and eh: lines.append(f"② 大盘×买入区:当前价{'在买入区' if el <= price <= eh else ('低于买入区' if price < el else '高于买入区')}{el}~{eh}") lines.append(f"③ 大盘×资金流:信号={signal},分类={category}") if sector: lines.append(f"④ 行业×基本面:{sector}") if tech: # 提取形态和量价 shape_m = re.search(r'形态:([^\s]+)', tech) vol_m = re.search(r'量价:([^\s]+)', tech) shape = shape_m.group(1) if shape_m else "?" vol = vol_m.group(1) if vol_m else "?" lines.append(f"⑤ 个股×技术面:{shape} | {vol}") sr_m = re.search(r'强撑:([\d.]+).*?弱撑:([\d.]+).*?弱压:([\d.]+).*?强压:([\d.]+)', tech) if sr_m: lines.append(f"⑥ 个股×支撑阻力:强撑{sr_m.group(1)}→弱撑{sr_m.group(2)}→弱压{sr_m.group(3)}→强压{sr_m.group(4)}") if sl and tp: lines.append(f"⑦ 个股×风报比:止损{sl} 止盈{tp} RR={rr:.1f}") if act: lines.append(f"⑧ 策略判定:{act[:100]}") lines.append(f"⑨ 综合结论:{signal},建议{'关注买入' if '买入' in signal else '持有观望' if signal == '持有' else '观望等待'}") return "\n".join(lines) def main(): codes = [a for a in sys.argv[1:] if not a.startswith("-")] if not codes: print("[FULL] 无指定编码,跑全量 regenerate_all()") from strategy_lifecycle import regenerate_all regenerate_all(stdout=False) print("[FULL] 全量重评完成") return # 读现有 decisions raw = read_decisions() decisions_map = {d["code"]: d for d in raw.get("decisions", []) if d.get("code")} ok = 0 errors = 0 skipped = 0 for code in codes: entry = decisions_map.get(code) if not entry: # 不在 decisions 中的自选股 → 从 DB watchlist_stocks 构建entry import sqlite3 _db = sqlite3.connect('/home/hmo/web-dashboard/data/mofin.db') _db.row_factory = sqlite3.Row _wl = _db.execute("SELECT * FROM watchlist_stocks WHERE code=? AND is_active=1", (code,)).fetchone() _db.close() if _wl: entry = { "code": code, "name": _wl["name"], "price": _wl["price"] or 0, "cost": 0, "shares": 0, "entry_low": _wl["entry_low"] or 0, "entry_high": _wl["entry_high"] or 0, "stop_loss": _wl["stop_loss"] or 0, "take_profit": 0, "action": "", "type": "自选策略", "is_watchlist": True, "analysis": json.loads(_wl["analysis_json"]) if _wl["analysis_json"] else {} } print(f"[WL] {code} {_wl['name']}: 从自选表构建entry") if not entry: print(f"[SKIP] {code}: 不在 decisions 或 watchlist_stocks 中") errors += 1 continue try: # Always fetch live price for accurate reassessment price = 0 try: # 价格从 DB 读取(price_monitor 每2分钟更新,唯一价格入口) code_raw = entry.get("code", "") price = 0 import sqlite3 db = sqlite3.connect('/home/hmo/web-dashboard/data/mofin.db') db.row_factory = sqlite3.Row row = db.execute("SELECT price FROM holdings WHERE code=? AND is_active=1", (code_raw,)).fetchone() if not row: row = db.execute("SELECT price FROM watchlist_stocks WHERE code=? AND is_active=1", (code_raw,)).fetchone() if not row: row = db.execute("SELECT price FROM holding_strategies WHERE code=? AND status='active' ORDER BY updated_at DESC LIMIT 1", (code_raw,)).fetchone() if row: price = row['price'] or 0 db.close() if price > 0: print(f" 实时价: {price} (来自DB)") else: # fallback to DB portfolio data _pf_data = read_portfolio() for _h in _pf_data.get("holdings", []): if _h["code"] == code_raw: price = float(_h.get("price", 0)) break if price <= 0: price = entry.get("current_price") or entry.get("price") or 0 except Exception as e: print(f" 价格获取失败: {e}", file=sys.stderr) price = entry.get("current_price") or entry.get("price") or 0 # Price diff debounce: skip reassessment if price changed < 1% since last update last_price = entry.get("last_reassessed_price") or 0 if last_price > 0 and price > 0: diff_pct = abs(price - last_price) / last_price * 100 if diff_pct < 1.0: print(f" 价差仅{diff_pct:.2f}% (<1%),跳过重评(上次价={last_price},现价={price})") skipped += 1 continue # 打印参数调试 print(f" DEBUG: code={code} name={entry.get('name','')} price={price} cost={entry.get('cost')} shares={entry.get('shares')} action={entry.get('action','')[:30]} is_wl={entry.get('type','') in ('自选策略','watchlist')}", flush=True) result = reassess_strategy( code=code, name=entry.get("name", ""), price=price or 0, cost=entry.get("cost") or 0, shares=entry.get("shares") or 0, current_action=entry.get("action", ""), is_watchlist=entry.get("type", "") in ("自选策略", "watchlist"), ) if result and result.get("action"): # 持仓股止损不下移(移动止损规则):已有仓位的止损只上不下 is_held = (entry.get("cost") or 0) > 0 and (entry.get("shares") or 0) > 0 and \ entry.get("type", "") not in ("自选策略", "watchlist") old_stop = entry.get("stop_loss") or 0 new_stop = result.get("stop_loss") or 0 if is_held and old_stop > 0 and new_stop > 0 and new_stop < old_stop: print(f" 移动止损保护: {new_stop}→保持{old_stop} (持仓止损不下移)") result["stop_loss"] = old_stop # 同时更新 action 字符串中的止损值 act = result.get("action", "") if act: act = re.sub(r'止损[\d.]+', f'止损{old_stop}', act) result["action"] = act # ── 写入 DB holding_strategies 表(替代 decisions.json)── try: from mofin_db import get_conn, write_holding_strategy _conn = get_conn() _db_entry = { "code": code, "name": entry.get("name", ""), "price": price, "cost": entry.get("cost", 0), "shares": entry.get("shares", 0), "stop_loss": result.get("stop_loss", entry.get("stop_loss")), "take_profit": result.get("take_profit", entry.get("take_profit")), "entry_low": result.get("entry_low", entry.get("entry_low")), "entry_high": result.get("entry_high", entry.get("entry_high")), "currency": "HKD" if (len(str(code)) == 5 and str(code)[0] in '01') else "CNY", "strategy_type": "自选策略" if entry.get("type", "") in ("自选策略", "watchlist") else "持仓策略", "action": result.get("action", ""), "timing_signal": result.get("timing_signal", entry.get("timing_signal", "")), "rr_ratio": result.get("rr_ratio", entry.get("rr_ratio", 0)), "tech_snapshot": result.get("tech_snapshot", entry.get("tech_snapshot", "")), "stock_category": result.get("stock_category", entry.get("stock_category", "")), "sector_context": result.get("sector_context", entry.get("sector_context", "")), "status": result.get("status", "active"), "source": entry.get("source", "auto"), "reason": result.get("action_note", ""), "version": entry.get("version", 1), "full_analysis": _build_full_analysis(code, entry, result) if result else "", } write_holding_strategy(_conn, code, entry.get("name", ""), _db_entry) _conn.commit() _conn.close() # 验证写入 _fa_check = _db_entry.get("full_analysis", "") print(f" DEBUG: full_analysis长度={len(_fa_check)} 内容=[{_fa_check[:100]}]") # 直接用SQL写入full_analysis try: _fa_conn = __import__('sqlite3').connect("/home/hmo/MoFin/data/mofin.db") _fa_conn.execute("UPDATE holding_strategies SET full_analysis=? WHERE code=? AND status='active'", (_fa_check, code)) _fa_conn.commit() _fa_conn.close() print(f" ✅ full_analysis直接SQL写入成功") except Exception as _fa_e: print(f" ⚠️ 直接SQL写入失败: {_fa_e}") _v = __import__('sqlite3').connect(str(__import__('pathlib').Path("/home/hmo/MoFin/data/mofin.db"))) _fa = _v.execute("SELECT full_analysis FROM holding_strategies WHERE code=? AND status='active'", (code,)).fetchone() if _fa and _fa[0]: print(f" ✅ full_analysis已写入({len(_fa[0])}字)") else: print(f" ⚠️ full_analysis为空") _v.close() # 用代码构建完整九维分析 _full_analysis_text = _build_full_analysis(code, entry, result) # 保存到DB _fa_conn = __import__('sqlite3').connect("/home/hmo/MoFin/data/mofin.db") _fa_conn.execute("UPDATE holding_strategies SET full_analysis=? WHERE code=? AND status='active'", (_full_analysis_text, code)) _fa_conn.commit() _fa_conn.close() print(f" ✅ 完整九维分析已保存({len(_full_analysis_text)}字)") print(f" [DB] holding_strategies 已更新: {code}") except Exception as _dbe: print(f" [DB FAIL] holding_strategies 写入失败: {_dbe}", file=sys.stderr) # 更新 decisions_map 中对应的条目 updated = entry.copy() # 币种标记:HK股保留HKD原始值,A股为CNY is_hk = len(str(code)) == 5 and str(code)[0] in '01' updated.update({ "action": result["action"], "stop_loss": result.get("stop_loss", entry.get("stop_loss")), "entry_low": result.get("entry_low", entry.get("entry_low")), "entry_high": result.get("entry_high", entry.get("entry_high")), "take_profit": result.get("take_profit"), "tech_snapshot": result.get("tech_snapshot", entry.get("tech_snapshot")), "timing_signal": result.get("timing_signal", entry.get("timing_signal")), "rr_ratio": result.get("rr_ratio", entry.get("rr_ratio", 0)), "status": result.get("status", "updated"), "price": price, "currency": "HKD" if is_hk else "CNY", }) # Save last reassessed price for debounce tracking updated["last_reassessed_price"] = price decisions_map[code] = updated # ——— 初始化多分支策略树 ——— try: sys.path.insert(0, '/home/hmo/MoFin') from strategy_tree import init_default_branches branches = init_default_branches( code, entry.get('name', ''), result.get('entry_low', 0), result.get('entry_high', 0), result.get('stop_loss', 0), result.get('take_profit', 0), ) st = updated.setdefault('strategy_tree', {}) st['branches'] = branches except Exception: pass print(f"[OK] {code} {entry.get('name','')}: {result['action'][:80]}") ok += 1 else: print(f"[SYNCED] {code}: 无变更") ok += 1 except Exception as e: print(f"[ERROR] {code}: {e}", file=sys.stderr) import traceback traceback.print_exc(file=sys.stderr) errors += 1 # 同步自选股更新回 watchlist_stocks 表(持仓策略已通过 write_holding_strategy 写入 DB) try: from datetime import datetime as _dt import sqlite3 _db2 = sqlite3.connect('/home/hmo/web-dashboard/data/mofin.db') for _code in codes: _entry = decisions_map.get(_code) if _entry and _entry.get("is_watchlist"): _db2.execute(""" UPDATE watchlist_stocks SET entry_low=?, entry_high=?, stop_loss=?, price=?, analysis_json=json(?) WHERE code=? AND is_active=1 """, ( _entry.get("entry_low", 0), _entry.get("entry_high", 0), _entry.get("stop_loss", 0), _entry.get("price", 0), json.dumps({ "action": _entry.get("action",""), "take_profit": _entry.get("take_profit", 0), "stop_loss": _entry.get("stop_loss", 0), "tech_snapshot": _entry.get("tech_snapshot", ""), "rr": _entry.get("rr_ratio", 0), "reassessed_at": _dt.now().strftime("%Y-%m-%d") }, ensure_ascii=False), _code )) _db2.commit() _db2.close() if any(e.get("is_watchlist") for e in [decisions_map.get(c) for c in codes] if e): print("[SYNC] 自选股策略已同步回 watchlist_stocks 表") except Exception as e: print(f"[SYNC FAIL] watchlist_stocks 同步失败: {e}", file=sys.stderr) print(f"[DONE] {ok}成功 {skipped}跳过 {errors}失败") # ── 第二步:扫描自选股(watchlist),价格偏离买入区>20%触发重评 ── scan_watchlist_stocks() # ════════════════════════════════════════════════════════════════════ # 自选股扫描 # ════════════════════════════════════════════════════════════════════ def scan_watchlist_stocks(): """扫描自选股表 (watchlist_stocks),对价格偏离买入区 >20% 的股票自动重评。 偏离公式: max(|price - entry_low|, |price - entry_high|) / entry_low * 100 > 20 通过 technical_analysis.full_analysis() 获取最新支撑/阻力位, 更新 entry_low / entry_high / stop_loss / price / analysis_json。 每轮最多处理 3 只,超过时标记剩余数量待下次扫描。 """ import sqlite3, json from datetime import datetime from technical_analysis import full_analysis from mo_models import is_hk_stock DB = '/home/hmo/web-dashboard/data/mofin.db' db = sqlite3.connect(DB) db.row_factory = sqlite3.Row rows = db.execute( "SELECT * FROM watchlist_stocks WHERE is_active=1" ).fetchall() if not rows: print("[WL-SCAN] 自选股表为空,跳过") db.close() return # ── 筛选偏离 >20% 的股票 ── candidates = [] # (code, name, price, entry_low, entry_high, stop_loss, deviation, analysis_json) for r in rows: code = r["code"] name = r["name"] price = r["price"] or 0 entry_low = r["entry_low"] or 0 entry_high = r["entry_high"] or 0 stop_loss = r["stop_loss"] or 0 analysis_json = r["analysis_json"] if entry_low <= 0 or price <= 0: continue dev_low = abs(price - entry_low) dev_high = abs(price - entry_high) deviation = max(dev_low, dev_high) / entry_low * 100 if deviation > 20: candidates.append((code, name, price, entry_low, entry_high, stop_loss, deviation, analysis_json)) total_needed = len(candidates) print(f"[WL-SCAN] 自选股共{len(rows)}只,偏离>20%需重评: {total_needed}只") MAX_PER_RUN = 3 to_process = candidates[:MAX_PER_RUN] remaining = max(0, total_needed - MAX_PER_RUN) if remaining > 0: print(f"[WL-SCAN] 本轮限{MAX_PER_RUN}只,剩余{remaining}只待下次扫描") if not to_process: print("[WL-SCAN] 无需重评") db.close() return ok = 0 errors = 0 for code, name, price, old_low, old_high, old_stop, deviation, old_analysis_json in to_process: print(f"[WL-REASSESS] {code} {name}: 偏离{deviation:.1f}%,触发重评") try: ta = full_analysis(code) if not ta or "error" in ta: print(f" [WARN] TA失败: {ta}") errors += 1 continue sr = ta.get("support_resistance", {}) if "error" in sr: print(f" [WARN] 支撑/阻力计算失败: {sr}") errors += 1 continue new_price = ta.get("quote", {}).get("price", price) new_entry_low = round(sr.get("weak_support", old_low), 2) new_entry_high = round(sr.get("weak_resist", old_high), 2) new_stop_loss = round(sr.get("strong_support", old_stop), 2) new_take_profit = round(sr.get("strong_resist", 0), 2) # ── 更新 analysis_json + changelog ── old_analysis = json.loads(old_analysis_json) if old_analysis_json else {} changelog = old_analysis.get("changelog", []) changelog.append({ "action": "auto_reassess_watchlist", "reason": f"价格偏离买入区{deviation:.1f}%", "old_entry_low": old_low, "old_entry_high": old_high, "new_entry_low": new_entry_low, "new_entry_high": new_entry_high, "old_stop_loss": old_stop, "new_stop_loss": new_stop_loss, "take_profit": new_take_profit, "price": new_price, "deviation_pct": round(deviation, 1), "timestamp": datetime.now().strftime("%Y-%m-%d %H:%M"), }) new_analysis = { **old_analysis, "take_profit": new_take_profit, "tech_snapshot": { "support_resistance": sr, "candlestick": ta.get("candlestick", {}), "volume": ta.get("volume", {}), "analyzed_at": ta.get("analyzed_at", ""), }, "reassessed_at": datetime.now().strftime("%Y-%m-%d"), "changelog": changelog, } currency = "HKD" if is_hk_stock(code) else "CNY" db.execute(""" UPDATE watchlist_stocks SET entry_low=?, entry_high=?, stop_loss=?, price=?, currency=?, analysis_json=? WHERE code=? AND is_active=1 """, ( new_entry_low, new_entry_high, new_stop_loss, new_price, currency, json.dumps(new_analysis, ensure_ascii=False), code, )) db.commit() print(f" [OK] {code} {name}: 买入区{old_low}-{old_high} -> {new_entry_low}-{new_entry_high}, " f"止损{new_stop_loss}, 止盈{new_take_profit}") ok += 1 except Exception as e: import traceback print(f" [ERROR] {code}: {e}", file=sys.stderr) traceback.print_exc(file=sys.stderr) errors += 1 db.close() remaining_msg = f" (剩余{remaining}只)" if remaining else "" print(f"[WL-SCAN] DONE: {ok}成功 {errors}失败{remaining_msg}") if __name__ == "__main__": main()