#!/usr/bin/env python3 """ per_stock_reassess.py — 按个股触发重评 对每只传进来的 code 执行 reassess_with_context()，然后写入 DB holding_strategies 表（纯DB模式，已移除JSON依赖）。 """ import sys, json, os, re from datetime import datetime COOLDOWN_HOURS_TRADING = 1 # 交易时段冷却（1小时） COOLDOWN_HOURS_NONTRADING = 24 # 非交易时段冷却 def _in_cooldown(code): """检查个股是否在重评冷却期内""" try: import sqlite3 conn = sqlite3.connect("/home/hmo/MoFin/data/mofin.db") r = conn.execute("SELECT reassessed_at FROM holding_strategies WHERE code=? AND status='active' ORDER BY id DESC LIMIT 1", (code,)).fetchone() conn.close() if not r or not r[0]: return False # 从未重评，立即执行 last = datetime.fromisoformat(r[0]) now = datetime.now() # 交易时段 vs 非交易时段 if 9 <= now.hour < 15: hours = COOLDOWN_HOURS_TRADING else: hours = COOLDOWN_HOURS_NONTRADING diff = (now - last).total_seconds() / 3600 return diff < hours except: return False sys.path.insert(0, "/home/hmo/web-dashboard") sys.path.insert(0, "/home/hmo/MoFin") sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) # profile-scripts 硬链目录 from strategy_lifecycle import reassess_with_context as reassess_strategy from mo_data import read_decisions, read_portfolio from llm_client import call_llm, REASSESS_MODEL from mofin_db import snapshot_strategy_history # ── 消息通道统一路由(broadcast/xmpp by delivery) ── try: from messenger import install_stdio_hook as _msh _msh() except Exception: pass def _build_full_analysis(code, entry, result): """从重评结果构建完整12维分析文本""" 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", "") # 2026-08-18 修复：entry_low/high 不用 or 短路——LLM 显式给（含0空区间）就用 LLM 值 # 否则 `0 or 95.0` 会把 LLM 的"清空区间"误判为"取旧脏值"（600262 教训：95/99 残留） # 2026-08-18 修复：entry_low/high 用 holding 值作为基础，LLM 算出区间（>0）才覆盖 # LLM 给 0（没算出）时用 holding 的合理值（不清空）；holding 是脏值时已被 promote 可执行性检查拦住 el = entry.get("entry_low", 0) if result.get("entry_low") and result.get("entry_low") > 0: el = result.get("entry_low") eh = entry.get("entry_high", 0) if result.get("entry_high") and result.get("entry_high") > 0: eh = result.get("entry_high") sl = result.get("stop_loss") or entry.get("stop_loss", 0) tp = result.get("take_profit") or entry.get("take_profit", 0) # 2026-08-18 修复：删除 rr_ratio 覆盖——RR 由 scanner 定（candidates.rr=holding.rr_ratio）， # per_stock_reassess 不该重算/覆盖（它该用 holding 的 rr_ratio，不重算）。 # 正确架构：RR 是 scanner 定义的单一事实来源，per_stock_reassess 只读不重算。 rr = entry.get("rr_ratio", 0) act = result.get("action", "") # ── 从DB拉取大盘、基本面、资金流 ── macro_desc = "" pe_val = pb_val = "" try: import sqlite3 as _sq, json as _j _db = _sq.connect("/home/hmo/MoFin/data/mofin.db") # 大盘（从structure列读取） _m = _db.execute("SELECT structure, sector_mood FROM macro_context_log ORDER BY id DESC LIMIT 1").fetchone() if _m and _m[0]: _st = _j.loads(_m[0]) _ix = _st.get("indices", {}) _desc = _st.get("description", "") if _ix: _parts = [] for _name in ["上证指数", "深证成指", "创业板指", "科创50", "恒生指数"]: if _name in _ix: _d = _ix[_name] if isinstance(_d, dict): _p = _d.get("price", 0) _c = _d.get("change_pct", 0) _parts.append(f"{_name}({_p:.0f},{_c:+.1f}%)") elif isinstance(_d, (int, float)): _parts.append(f"{_name}({_d})") macro_desc = " ".join(_parts) elif _desc: macro_desc = _desc _mood = str(_m[1] or "") if _mood and not macro_desc: macro_desc = f"情绪={_mood}" elif _mood: macro_desc += f" 情绪={_mood}" if not macro_desc: # fallback: 直接用腾讯API拉大盘 try: _r2 = __import__('subprocess').run(["curl", "-s", "http://qt.gtimg.cn/q=sh000001,sz399001,sz399006,sh000688"], capture_output=True, timeout=10) _txt = _r2.stdout.decode("gbk", errors="ignore") _parts = [] for _line in _txt.strip().split("\n"): if "~" not in _line: continue _p = _line.split("~") if len(_p) < 4: continue _name2 = _p[1] _price2 = _p[3] _chg2 = _p[32] if len(_p) > 32 else "0" _parts.append(f"{_name2}({_price2},{_chg2}%)") if _parts: macro_desc = "腾讯实时 " + " ".join(_parts[:3]) except: pass # 基本面+实时价：直接从腾讯API拉（盘后也有收盘价） try: _pfx = "sh" if str(code).startswith(("6", "9")) else "sz" _r3 = __import__('subprocess').run(["curl", "-s", f"http://qt.gtimg.cn/q={_pfx}{code}"], capture_output=True, timeout=10) _txt3 = _r3.stdout.decode("gbk", errors="ignore") _p3 = _txt3.split("~") if len(_p3) > 45: _pe = _p3[39] if _p3[39] else "" _pb = _p3[40] if len(_p3) > 40 and _p3[40] else "" _mcap = _p3[44] if len(_p3) > 44 and _p3[44] else "" _price_now = float(_p3[3]) if _p3[3] else 0 _chg_now = float(_p3[32]) if len(_p3) > 32 and _p3[32] else 0 if _price_now > 0: price = _price_now # 覆盖策略中的price=0 if _pe: pe_val = f"PE={_pe}" if _pb: pb_val = f"PB={_pb}" if _mcap: mcap_val = f"市值{float(_mcap)/10000:.1f}亿" if float(_mcap) > 10000 else f"市值{_mcap}万" pe_val += f" {mcap_val}" if pe_val else mcap_val except: pass _db.close() except Exception as _e: pass # ── 从tech_snapshot提取MA和支撑阻力 ── import re ma5 = ma10 = ma20 = ma60 = "?" 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} 12维全析)】") lines.append("") if macro_desc: lines.append(f"① 大盘环境（当日实时）：{macro_desc}") else: lines.append(f"① 大盘环境（当日实时）：数据待刷新") if pe_val or pb_val: lines.append(f"② 个股基本面（最新财报）：{pe_val} {pb_val}") else: lines.append(f"② 个股基本面（最新财报）：数据待补充") lines.append(f"③ 技术面（MA5/10/20/60日 支撑阻力近20日）：MA5={ma5} MA10={ma10} MA20={ma20} MA60={ma60}") if el and eh and price > 0: pos = "在买入区内" if el <= price <= eh else (f"低于买入区{(1-price/el)*100:.0f}%" if price < el else f"高于买入区{(price/eh-1)*100:.0f}%") lines.append(f"④ 价格位置：{price} {pos} 区间{el}~{eh}") else: lines.append(f"④ 价格位置：数据待刷新") if sl and tp and rr: lines.append(f"⑤ 风报比：止损{sl} 止盈{tp} RR={rr:.1f}") # 支撑阻力 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 sector: lines.append(f"⑦ 行业背景：{sector}") else: # 从stock_sectors表补行业 try: _s2 = __import__('sqlite3').connect("/home/hmo/MoFin/data/mofin.db") _sr = _s2.execute("SELECT sector_name FROM stock_sectors WHERE code=? LIMIT 1", (code,)).fetchone() if _sr and _sr[0]: lines.append(f"⑦ 行业背景：{_sr[0]}") else: # 2026-08-17 修复：stock_sectors 仅898只覆盖不全，改读 stock_sectors_em（5061只） _sr2 = _s2.execute("SELECT sector FROM stock_sectors_em WHERE code=? LIMIT 1", (code,)).fetchone() if _sr2 and _sr2[0]: lines.append(f"⑦ 行业背景：{_sr2[0]}") _s2.close() except: pass # 消息面：从signal_news读最新信号（不限情绪标签，LLM自行判断） news_lines = [] try: _n_db = __import__('sqlite3').connect("/home/hmo/MoFin/data/mofin.db") _nr = _n_db.execute( "SELECT summary, overall_sentiment, created_at FROM signal_news " "WHERE sector LIKE ? OR sector LIKE ? " "ORDER BY id DESC LIMIT 2", (f'%{code}%', f'%{name[:4]}%') ).fetchall() if not _nr: _nr = _n_db.execute( "SELECT summary, overall_sentiment, created_at FROM signal_news " "ORDER BY id DESC LIMIT 2").fetchall() for _ns in _nr: _sent = str(_ns[1]) if '利好' in _sent: _icon = '📈' elif '利空' in _sent: _icon = '📉' else: _icon = '📰' news_lines.append(f"{_icon} {_ns[0][:60]} ({str(_ns[2])[:10]})") _n_db.close() except: pass if category: lines.append(f"⑧ 分类评级：{category}") lines.append(f"⑨ 策略信号：{signal}") if news_lines: lines.append("") lines.extend(news_lines) if act: lines.append(f"\n策略详情：{act[:200]}") return "\n".join(lines) def main(): codes = [a for a in sys.argv[1:] if not a.startswith("-")] if not codes: # 2026-08-14 修复：无参数时不再跑全量 regenerate_all（那是盘前 premarket 的职责，超时 600s） # 改为只处理盘中到期的自选（scan_watchlist_stocks，MAX_PER_RUN=3 限制，快） print("[WL-SCAN] 无指定编码，扫描盘中到期自选（不跑全量 regenerate_all）") scan_watchlist_stocks() 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: # 冷却期检查 if _in_cooldown(code): print(f" ⏭ {code}: 冷却期内跳过") skipped += 1 continue entry = decisions_map.get(code) if not entry: # 不在 decisions 中的自选股 → 从 holding_strategies 构建entry import sqlite3 _db = sqlite3.connect('/home/hmo/MoFin/data/mofin.db') _db.row_factory = sqlite3.Row _wl = _db.execute("SELECT * FROM holding_strategies WHERE code=? AND status='active' AND decision_type='自选策略'", (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: # 2026-08-18 修复：三表无价时 stock_quote 直查（600262 案例：promote 后未进价格表→price=0→计算全崩） try: import subprocess, json as _jj2 _rq = subprocess.run(["python3", "/home/hmo/.hermes/profiles/position-analyst/scripts/stock_quote.py", code_raw], capture_output=True, text=True, timeout=10) _qq = _jj2.loads(_rq.stdout) _qprice = float(_qq.get("price", 0)) if _qprice > 0: price = _qprice print(f" 实时价: {price} (stock_quote直查)") except Exception as _qe: print(f" stock_quote直查失败: {_qe}", file=sys.stderr) if price <= 0: # 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 # 打印参数调试 if entry is None: print(f" DEBUG: code={code} ENTRY=NONE 跳过") print(f" [SKIP] {code} 策略数据不存在") skipped += 1 continue entry_action = str(entry.get('action') or '') print(f" DEBUG: code={code} name={entry.get('name','')} price={price} cost={entry.get('cost')} shares={entry.get('shares')} action={entry_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": entry.get("rr_ratio", 0), # 2026-08-18 RR由scanner定,不覆盖(用holding的) "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, source_trigger="per_stock_12d") # 2026-08-18 LLM路径不被技术参数保护 _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() # LLM生成完整12维分析（2026-07-23：统一走 batch 的 collect_data+build_prompt。 # 单一 prompt 源头，根治双 prompt 漂移——per_stock 曾缺技术位锚/持仓上下文/参数自检） import sys as _sys2 if '/home/hmo/MoFin/deploy/profile-scripts' not in _sys2.path: _sys2.path.insert(0, '/home/hmo/MoFin/deploy/profile-scripts') try: from batch_reassess import collect_data as _cd, build_prompt as _bp _prompt = _bp(_cd(code)) except Exception as _pe: print(f" ⚠️ 统一prompt构建失败: {_pe}", flush=True) _prompt = None _full_analysis_text = None if _prompt: try: # 2026-08-13 超时修复：timeout 150→90, retries 1→0（3只×(90+90+20)=600s临界,减到90+0重试=270s安全） _llm_result = call_llm(_prompt, max_tokens=None, timeout=300, retries=0, backoff=0, concurrent=True) # 2026-08-24 并发模式: router round-robin分key(6 key齐用),90→300防自断 if _llm_result["ok"]: _full_analysis_text = _llm_result["content"] print(f" ✅ LLM12维分析完成({len(_full_analysis_text)}字, {_llm_result['elapsed']:.1f}s)", flush=True) else: print(f" ❌ LLM12维分析失败({_llm_result['attempts']}次): {_llm_result['error'][:200]}", flush=True) except Exception as _e: print(f" ❌ LLM12维分析异常: {_e}", flush=True) # ── 保存到DB（覆写前先快照）── _fa_conn = __import__('sqlite3').connect("/home/hmo/MoFin/data/mofin.db") if _full_analysis_text: # 快照旧策略（使用共享函数） try: snapshot_strategy_history(_fa_conn, code, "per_stock_12d") except Exception as _se: print(f" ⚠️ 快照失败: {_se}", flush=True) _fa_conn.execute( "UPDATE holding_strategies SET full_analysis=?, reassessed_at=? WHERE code=? AND status='active'", (_full_analysis_text, __import__('datetime').datetime.now().isoformat(), code)) _fa_conn.commit() _fa_conn.close() if _full_analysis_text: print(f" ✅ 完整12维分析已保存({len(_full_analysis_text)}字)") else: print(f" ⚠️ 12维分析未完成，跳过保存") print(f" [DB] holding_strategies 已更新: {code}") # 信号以已存分析为唯一事实源（防信号/分析脱节） # 推荐推送统一走 reconcile→tag→摘要队列（batch 结束统一发，不再单只推送） from mofin_db import reconcile_signal_from_analysis _rc_conn = __import__('sqlite3').connect('/home/hmo/MoFin/data/mofin.db') _sig = reconcile_signal_from_analysis(_rc_conn, code) _rc_conn.close() if _sig: print(f" ✅ LLM信号={_sig} 已对齐") # 2026-08-18 策略判断落库（复用 batch 的 parse_response 从 full_analysis 提取） try: from batch_reassess import parse_response as _pr2 _p2 = _pr2(_full_analysis_text or "") _j2 = _p2.get("strategy_judge", "") _st2 = _p2.get("strategy_switch_to", "") if _j2 in ("策略失效需更换", "策略失效重定"): if _st2: _rc_conn = __import__('sqlite3').connect('/home/hmo/MoFin/data/mofin.db') _rc_conn.execute("UPDATE holding_strategies SET strategy_attributed=?, strategy_state='switched', strategy_provenance='llm_attributed' WHERE code=? AND status='active'", (_st2, code)) _rc_conn.commit(); _rc_conn.close() print(f" ✅ 策略切换: {_j2} → 归属 {_st2}（switched）") else: _rc_conn = __import__('sqlite3').connect('/home/hmo/MoFin/data/mofin.db') _rc_conn.execute("UPDATE holding_strategies SET strategy_state='invalidated' WHERE code=? AND status='active'", (code,)) _rc_conn.commit(); _rc_conn.close() print(f" ✅ 策略失效: {_j2} 无合适归属（invalidated）") elif _j2: print(f" ℹ️ 策略判断={_j2}（维持/修改参数，不落库）") except Exception as _pe: print(f" ⚠️ 策略判断落库失败: {_pe}") # 冷却期已更新（reassessed_at写入） 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": entry.get("rr_ratio", 0), # 2026-08-18 RR由scanner定不覆盖(用holding的) "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}失败") # ── 推荐摘要发货（per_stock 路径产生的推荐也要出队列）── try: from mofin_db import flush_rec_digest flush_rec_digest() except Exception as _fe: print(f" ⚠️ 推荐摘要发送失败: {_fe}", flush=True) # ── 第二步：扫描自选股（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()