From d28a32e24439f20340f65fa614f9be1e157e7beb Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E7=9F=A5=E5=BE=AE?= Date: Mon, 20 Jul 2026 17:47:29 +0800 Subject: [PATCH] chore: deployed price_events dual-write --- deploy/profile-scripts/price_monitor.py | 1504 ++++++++++++----------- 1 file changed, 762 insertions(+), 742 deletions(-) diff --git a/deploy/profile-scripts/price_monitor.py b/deploy/profile-scripts/price_monitor.py index 5d23c525..1fb2fb0f 100644 --- a/deploy/profile-scripts/price_monitor.py +++ b/deploy/profile-scripts/price_monitor.py @@ -1,742 +1,762 @@ -#!/usr/bin/env python3 -"""price_monitor.py — 高频价格监控脚本（批量版） -规则：进入区间报一次，离开区间报一次，中间不重复。 -每次运行时一次性刷新所有持仓+自选股的实时价。 -""" -import urllib.request -import os, sys, time, json -import sqlite3 -from datetime import datetime - -from mo_data import read_decisions - -BREACH_PATH = "/home/hmo/.hermes/zone_breach.json" -STATE_PATH = "/home/hmo/.hermes/price_trigger_state.json" -EVENTS_PATH = "/home/hmo/web-dashboard/data/price_events.json" - -# DB 模块（同步实时价到 mofin.db） -sys.path.insert(0, "/home/hmo/MoFin") -try: - from mofin_db import get_conn, DB_PATH - from mo_models import calc_total_mv, calc_total_assets - HAS_DB = True -except ImportError: - HAS_DB = False - -# 策略重评依赖（技术面驱动，非机械百分比） -sys.path.insert(0, "/home/hmo/web-dashboard") -try: - from strategy_lifecycle import reassess_strategy, reassess_with_context - HAS_REASSESS = True -except ImportError: - HAS_REASSESS = False - -UA = "Mozilla/5.0" - -# ── XMPP推送 ────────────────────────────────────────────────────────── -XMPP_USER = "hmo@yoin.fun" -XMPP_BRIDGE = "http://127.0.0.1:5805/" - -def push_to_xmpp(text): - """通过知微 HTTP bridge 推送到Dad私信""" - if not text.strip(): - return - try: - payload = json.dumps({ - "to": XMPP_USER, - "body": text.strip(), - "type": "chat", - }).encode("utf-8") - req = urllib.request.Request(XMPP_BRIDGE, data=payload, headers={"Content-Type": "application/json"}) - urllib.request.urlopen(req, timeout=5) - except Exception as e: - print(f"[XMPP推送失败] {e}", file=sys.stderr) - -# ── 批量拉取价格 ────────────────────────────────────────────────────────── - -def fetch_all_prices(codes): - """腾讯批量行情API：一次请求拉取所有股票（A股+港股） - A股：sh600110 / sz000001 - 港股：hk00700 - 返回 {code: (price, change, change_pct)} - """ - if not codes: - return {} - - # 构建批量查询串 - symbols = [] - code_map = {} # symbol -> original_code - for code in codes: - code_s = str(code).strip() - if len(code_s) == 6: - # A股：沪市以5/6/9开头，深市以0/3开头 - if code_s.startswith(('5', '6', '9')): - sym = f"sh{code_s}" - else: - sym = f"sz{code_s}" - else: - sym = f"hk{code_s}" - symbols.append(sym) - code_map[sym] = code_s - - url = f"http://qt.gtimg.cn/q={','.join(symbols)}" - try: - req = urllib.request.Request(url, headers={"User-Agent": UA}) - with urllib.request.urlopen(req, timeout=10) as r: - text = r.read().decode("gbk") - except Exception as e: - print(f"⚠️ 批量拉取失败: {e}", file=sys.stderr) - return {} - - results = {} - for line in text.strip().split("\n"): - line = line.strip() - if not line or "=" not in line: - continue - try: - # 格式: v_sh600110="1~诺德股份~600110~11.84~11.90~..." - raw_value = line.split("=", 1)[1].strip().strip('"').strip(";") - fields = raw_value.split("~") - if len(fields) < 6: - continue - sym = line.split("=", 1)[0].strip().lstrip("v_") - orig_code = code_map.get(sym) - if not orig_code: - continue - price = float(fields[3]) if fields[3] else 0 - prev_close = float(fields[4]) if fields[4] else 0 - change = price - prev_close if prev_close > 0 else 0 - change_pct = fields[32] if len(fields) > 32 and fields[32] else "0" - results[orig_code] = (price, change, change_pct) - except (ValueError, IndexError): - continue - - return results - - -def refresh_data_prices(): - """一次性刷新所有持仓+自选股的实时价（完全DB版，不写JSON）""" - all_codes = set() - - # 从DB读所有需要拉取价格的代码 - try: - conn = get_conn() - for r in conn.execute("SELECT code FROM holdings WHERE is_active=1"): - all_codes.add(r['code']) - for r in conn.execute("SELECT code FROM watchlist_stocks"): - all_codes.add(r['code']) - for r in conn.execute("SELECT code FROM holding_strategies WHERE status='active'"): - all_codes.add(r['code']) - conn.close() - except Exception as e: - print(f"⚠️ 从DB读代码失败: {e}", file=sys.stderr) - return 0 - - if not all_codes: - return 0 - - # 一次性批量拉取 - prices = fetch_all_prices(list(all_codes)) - updated = len(prices) - - # === 弹性同步实时价到 mofin.db === - # 防死锁策略（经2026-07-14 WAL死锁复盘改进）： - # ① 启动时 checkpoint WAL（清理残留事务） - # ② 统一 BEGIN IMMEDIATE 包裹整个写操作 - # ③ 5次重试 + 指数退避: 1s → 2s → 4s → 8s → 16s（共~31s） - # ④ get_conn() 的 busy_timeout=30000 保证等待上限 - # ⑤ 每个写操作检查返回值，任一失败立即 rollback + 重试 - # ⑥ try/finally 确保连接始终释放 - if HAS_DB and prices: - # 先checkpoint一次，清理上次被kill残留的WAL - try: - c = get_conn() - c.execute("PRAGMA wal_checkpoint(TRUNCATE)") - c.close() - except Exception: - pass - - max_tries = 5 - conn = None - for db_attempt in range(max_tries): - try: - conn = get_conn() - # BEGIN IMMEDIATE 立即获取写锁——失败则等 busy_timeout(30s) - conn.execute("BEGIN IMMEDIATE") - - # ── 构建 holdings 更新数据 ── - db_holdings = [] - for r in conn.execute("SELECT * FROM holdings WHERE is_active=1"): - h = dict(r) - code = str(h.get('code', '')) - if code in prices: - price_val, _, change_pct = prices[code] - if price_val > 0: - h['price'] = round(price_val, 2) - h['change_pct'] = float(change_pct) if change_pct else 0 - db_holdings.append(h) - - # ── 写 holdings 表 ── - for h in db_holdings: - currency = str(h.get('currency', 'CNY')).upper() - if currency not in ('CNY', 'HKD'): - raise ValueError(f"非法币种: {currency}") - conn.execute(""" - INSERT INTO holdings (code, name, shares, cost, price, market_value, - change_pct, currency, position_pct, added_at, is_active) - VALUES (?,?,?,?,?,?,?,?,?,datetime('now','localtime'),1) - ON CONFLICT(code) DO UPDATE SET - name=excluded.name, shares=excluded.shares, cost=excluded.cost, - price=excluded.price, market_value=excluded.market_value, - change_pct=excluded.change_pct, currency=excluded.currency, - position_pct=excluded.position_pct - """, ( - h.get('code'), h.get('name'), h.get('shares', 0), - h.get('cost'), h.get('price'), - h.get('market_value'), h.get('change_pct'), - h.get('currency', 'CNY'), h.get('position_pct'), - )) - - # ── 写 portfolio_summary ── - mv = calc_total_mv(db_holdings) - existing = conn.execute( - 'SELECT cash, frozen_cash FROM portfolio_summary WHERE id=1' - ).fetchone() - db_cash = existing['cash'] if existing else 0.0 - db_frozen = existing['frozen_cash'] if existing else 0.0 - assets = calc_total_assets({'holdings': db_holdings, 'cash': db_cash, 'frozen_cash': db_frozen}) - position_pct = round(mv / assets * 100, 2) if assets > 0 else 0 - conn.execute(""" - INSERT INTO portfolio_summary (id, total_assets, total_mv, stock_value, - cash, frozen_cash, position_pct, total_pnl, currency, updated_at) - VALUES (1,?,?,?,?,?,?,?,?,datetime('now','localtime')) - ON CONFLICT(id) DO UPDATE SET - total_assets=excluded.total_assets, total_mv=excluded.total_mv, - stock_value=excluded.stock_value, cash=excluded.cash, - frozen_cash=excluded.frozen_cash, position_pct=excluded.position_pct, - total_pnl=excluded.total_pnl, currency=excluded.currency, - updated_at=datetime('now','localtime') - """, ( - assets, mv, mv, db_cash, db_frozen, - position_pct, 0, 'CNY', - )) - - # ── 写 live_prices ── - for h in db_holdings: - code = h.get('code', '') - if code: - p = h.get('price', 0) - cp = h.get('change_pct', 0) - conn.execute( - "INSERT OR REPLACE INTO live_prices (code, price, change_pct, updated_at) " - "VALUES (?,?,?,datetime('now','localtime'))", - (code, p, cp) - ) - # 补充策略股/自选股的价格（不在holdings中的） - for code, pdata in prices.items(): - if code not in {h.get('code') for h in db_holdings}: - price_val = pdata[0] if isinstance(pdata, (list, tuple)) else pdata.get('price', 0) - cp_val = pdata[1] if isinstance(pdata, (list, tuple)) else pdata.get('change_pct', 0) - conn.execute( - "INSERT OR REPLACE INTO live_prices (code, price, change_pct, updated_at) " - "VALUES (?,?,?,datetime('now','localtime'))", - (code, price_val, cp_val) - ) - - conn.commit() - conn.close() - conn = None - if db_attempt > 0: - print(f"DB同步成功（第{db_attempt+1}次重试）") - break # success - - except (sqlite3.OperationalError, sqlite3.DatabaseError) as e: - if conn: - try: conn.rollback() - except Exception: pass - try: conn.close() - except Exception: pass - conn = None - err_str = str(e) - if "locked" in err_str or "cannot commit" in err_str or "busy" in err_str: - if db_attempt < max_tries - 1: - wait = 2 ** db_attempt # 1, 2, 4, 8, 16 - print(f"⏳ DB锁(尝试{db_attempt+1}/{max_tries}): {e} → {wait}s后重试", file=sys.stderr) - time.sleep(wait) - else: - print(f"❌ DB锁(重试{max_tries}次耗尽): {e}", file=sys.stderr) - else: - print(f"❌ DB错误: {e}", file=sys.stderr) - break - except Exception as e: - if conn: - try: conn.rollback() - except Exception: pass - try: conn.close() - except Exception: pass - conn = None - print(f"⚠️ DB同步异常: {e}", file=sys.stderr) - break - else: - # for-else: loop exhausted without break - print("❌ DB同步失败（所有重试耗尽）", file=sys.stderr) - # 尝试紧急 WAL checkpoint（释放死锁） - try: - c = sqlite3.connect(str(DB_PATH), timeout=1) - c.execute("PRAGMA wal_checkpoint(TRUNCATE)") - c.close() - print(" ↪ 紧急WAL checkpoint完成", file=sys.stderr) - except Exception as we: - print(f" ↪ WAL checkpoint也失败: {we}", file=sys.stderr) - - return updated - - -# ── 区间偏离检测 ────────────────────────────────────────────────────────── - -def load_state(): - try: - with open(STATE_PATH) as f: - return json.load(f) - except: - return {} - -def save_state(state): - os.makedirs(os.path.dirname(STATE_PATH), exist_ok=True) - with open(STATE_PATH, 'w') as f: - json.dump(state, f, ensure_ascii=False, indent=2) - -def load_breaches(): - try: - with open(BREACH_PATH) as f: - return json.load(f) - except: - return {} - -def save_breaches(data): - os.makedirs(os.path.dirname(BREACH_PATH), exist_ok=True) - with open(BREACH_PATH, 'w') as f: - json.dump(data, f, ensure_ascii=False, indent=2) - - -def load_events(): - try: - with open(EVENTS_PATH) as f: - return json.load(f) - except: - return {"events": []} - - -def save_events(events): - os.makedirs(os.path.dirname(EVENTS_PATH), exist_ok=True) - with open(EVENTS_PATH, 'w') as f: - json.dump(events, f, ensure_ascii=False, indent=2) - - -def record_event(code, name, event_type, price, trigger_value, event_label=""): - """记录一次价格触发事件到 price_events.json""" - events = load_events() - now = datetime.now().isoformat() - events["events"].append({ - "code": code, - "name": name, - "event_type": event_type, # entry_zone, stop_loss, take_profit, exit_zone - "price": round(price, 2), - "trigger_value": trigger_value, - "event_label": event_label, - "timestamp": now, - "date": datetime.now().strftime("%Y-%m-%d"), - }) - # 保留最近10000条 - events["events"] = events["events"][-10000:] - save_events(events) - - -def get_trigger_zones(trigger): - """返回该trigger所有可监控的区间列表，跳过已执行的batch""" - zones = [] - for key, label in [ - ("entry_zone", "加仓区间"), - ("batch1_price", "试仓区间"), - ("batch2_price", "加仓区间"), - ("take_profit_zone", "止盈区间"), - ("watch_low", "关注区间"), - ("watch_high", "减仓区间"), - ("watch_break", "止损区间") - ]: - status_key = key.replace("_price", "_status") - if status_key in trigger and trigger[status_key] == "executed": - continue - val = trigger.get(key, "") - if val and "~" in val: - try: - parts = val.split("~") - lo, hi = float(parts[0]), float(parts[1]) - zones.append((key, label, lo, hi)) - except: - pass - sl = trigger.get("stop_loss", "") - if sl: - try: - sl_price = float(sl) if isinstance(sl, (int, float)) else float(sl) - zones.append(("stop_loss", "止损", 0, sl_price)) - except: - pass - return zones - - -def _cleanup_lock(): - """清理进程锁文件""" - try: - os.remove("/tmp/price_monitor.lock") - except Exception: - pass - -def _handle_sigterm(signum, frame): - """收到SIGTERM时清理锁文件后退出""" - _cleanup_lock() - sys.exit(0) - -def run_once(round_label=""): - """执行一轮完整的监控流程""" - import os, signal # 必须在开头import，否则os变量会被后面的局部import绑定覆盖 - signal.signal(signal.SIGTERM, _handle_sigterm) - os.nice(10) # 降低优先级，避免与DB其他写操作抢占 - # ── 进程锁：同一时间只跑一个实例 ── - _lk = "/tmp/price_monitor.lock" - _pid = None - try: - with open(_lk) as _f: - _pid = int(_f.read().strip()) - os.kill(_pid, 0) - print(f"[LOCK] 已有实例(PID {_pid})在运行，跳过本轮", file=sys.stderr, flush=True) - return - except (FileNotFoundError, ProcessLookupError, ValueError): - pass - with open(_lk, "w") as _f: - _f.write(str(os.getpid())) - - label = f" [{round_label}]" if round_label else "" - start = time.time() - TIME_BUDGET = 90 # 预留30s给输出和清理，90s内必须完成核心逻辑 - - # === 第一步：一次性刷新所有价格 === - refreshed = refresh_data_prices() - - # === 第二步：检查触发条件 === - try: - dec = read_decisions() - except: - print(f"❌{label} 无法读取decisions(DB)", file=sys.stderr) - return - - active = [d for d in dec.get("decisions", []) if d.get("status") == "active"] - state = load_state() - outputs = [] - state_updated = False - # 时间冷却：同股同区间30分钟内不重复推 - _push_cooldown = {} - _cooldown_file = "/home/hmo/.hermes/.price_push_cooldown.json" - try: - import os - if os.path.exists(_cooldown_file): - with open(_cooldown_file) as _f: - _push_cooldown = json.load(_f) - except Exception: - _push_cooldown = {} - - def _can_push(code, zone_key): - now = time.time() - key = f"{code}_{zone_key}" - last = _push_cooldown.get(key, 0) - if now - last < 1800: # 30分钟 - return False - _push_cooldown[key] = now - # 持久化写入 - try: - with open(_cooldown_file, "w") as _f: - json.dump(_push_cooldown, _f) - except Exception: - pass - return True - - # 收集所有需要检查的代码 - check_codes = set() - for d in active: - trig = d.get("trigger", {}) - if trig: - check_codes.add(d["code"]) - - # 批量拉取这些股票的价格 - prices = fetch_all_prices(list(check_codes)) - - for d in active: - code = d["code"] - trig = d.get("trigger", {}) - if not trig: - continue - - zones = get_trigger_zones(trig) - if not zones: - continue - - price_info = prices.get(code) - if not price_info: - continue - price, _, _ = price_info - if price == 0: - continue - - name = d.get("name", code) - if code not in state: - state[code] = {} - - # 时间预算检查：如果超时，跳过重评只做状态记录 - _budget_low = (time.time() - start) > TIME_BUDGET - - for key, label, lo, hi in zones: - in_zone = lo <= price <= hi - prev_in_zone = state[code].get(key, None) - - if in_zone and prev_in_zone != True: - if key == "stop_loss": - outputs.append(f"⚠️ {name}({code}) {price} → 跌破止损{hi}！") - record_event(code, name, "stop_loss", price, str(hi)) - # 止损触发 → 立即重评并推送给Dad（时间不够则直接推原始告警） - if _budget_low: - outputs.append(f" 📨 止损触发(超时跳过重评)→已推送Dad") - if _can_push(code, "stop_loss"): - push_to_xmpp(f"⚠️ {name}({code}) {price} → 跌破止损{hi}！") - else: - try: - cost = d.get("cost", 0) or 0 - shares = d.get("shares", 0) or 0 - current_action = d.get("action", "") - result = reassess_with_context(code, name, price, cost, shares, current_action) - if result: - timing_signal = result.get("timing_signal", "") - action = result.get("action", "") - if "买入" in timing_signal or "加仓" in timing_signal or timing_signal in ("卖出","止盈"): - buy_lo = d.get("entry_low", 0) - buy_hi = d.get("entry_high", 0) - rr = result.get("rr_ratio", 0) - if _can_push(code, "stop_loss"): - msg = f"🔔 {name}({code}) 价{price}→触发操作区间{max(buy_lo,0):.2f}~{buy_hi:.2f}，已触发重评|RR={rr}" - push_to_xmpp(msg) - outputs.append(f" 📨 止损重评→已推送Dad: {action}") - except Exception as e: - outputs.append(f" ⚠️ 止损重评失败: {e}") - else: - extra = "" - if "_price" in key: - batch_shares = trig.get(key.replace("_price", "_shares"), "") - action = trig.get(key.replace("_price", "_action"), "") - if batch_shares: - extra = f" {action}{batch_shares}股" if action else f" {batch_shares}股" - elif key in ("take_profit_zone",): - act = trig.get("take_profit_action", "") - if act: - extra = f"（{act}）" - outputs.append(f"⚡ {name}({code}) {price} → 进入{label}{lo}~{hi}{extra}") - record_event(code, name, "entry_zone", price, f"{lo}~{hi}", label) - # 进入区间 → 立即重评并推送给Dad（时间不够则跳过重评直接推原始告警） - if _budget_low: - if _can_push(code, key): - push_to_xmpp(f"⚡ {name}({code}) {price} → 进入{label}{lo}~{hi}") - outputs.append(f" 📨 区间触发(超时跳过重评)→已推送Dad") - else: - try: - cost = d.get("cost", 0) or 0 - shares = d.get("shares", 0) or 0 - current_action = d.get("action", "") - result = reassess_with_context(code, name, price, cost, shares, current_action) - if result: - timing_signal = result.get("timing_signal", "") - action = result.get("action", "") - # 格式化区间描述（止盈区lo=0时美化显示） - if key == "take_profit_zone" and lo == 0: - zone_desc = f"止盈监控(目标{hi:.0f})" - else: - zone_desc = f"操作区间{lo}~{hi}" - if "买入" in timing_signal or "加仓" in timing_signal or timing_signal in ("卖出","止盈"): - rr = result.get("rr_ratio", 0) - if _can_push(code, key): - msg = f"🔔 {name}({code}) 价{price}→触发{zone_desc}，已触发重评|RR={rr}" - push_to_xmpp(msg) - outputs.append(f" 📨 区间触发重评→已推送Dad: {action}") - else: - reason = f"重评结果:{timing_signal}，不构成操作建议" - outputs.append(f" 📋 本地日志(不推): {reason}") - except Exception as e: - outputs.append(f" ⚠️ 区间重评失败: {e}") - state[code][key] = True - state_updated = True - - elif not in_zone and prev_in_zone == True: - if key != "stop_loss": - outputs.append(f"📌 {name}({code}) {price} → 离开{label}{lo}~{hi}") - state[code][key] = False - state_updated = True - - # === 第三步：买入区偏离检测 + 自动重评 === - reassesed_codes = [] - # 先做急跌检测（仅持仓，自选股不推送暴跌告警） - holdings_codes = {d["code"] for d in active if (d.get("shares") or 0) > 0} - for d in active: - code = d["code"] - # 非持仓跳过 - if code not in holdings_codes: - continue - name = d.get("name", code) - price_info = prices.get(code) - if not price_info: - continue - price, _, change_pct = price_info - if price == 0: - continue - # 单日跌幅>7%告警（不依赖zone边界，盘中急跌即触发） - try: - cp = float(change_pct) if change_pct else 0 - except: - cp = 0 - if cp <= -7: - prev_alert = state.get(code, {}).get("__sharp_decline_triggered", False) - if not prev_alert: - stop_loss = d.get("stop_loss", 0) - sl_note = f" 止损{stop_loss}" if stop_loss else "" - msg = f"🔻 {name}({code}) {price} 暴跌{cp:.1f}%！{sl_note}" - push_to_xmpp(msg) - outputs.append(msg) - state.setdefault(code, {})["__sharp_decline_triggered"] = True - state_updated = True - # 立即持久化，防止后续超时导致状态丢失而重复推送 - save_state(state) - elif cp > -5: - # 反弹后清除告警标记，下次再跌还能报 - state.setdefault(code, {}).pop("__sharp_decline_triggered", None) - - for d in active: - code = d["code"] - name = d.get("name", code) - price_info = prices.get(code) - if not price_info: - continue - price, _, _ = price_info - if price == 0: - continue - - # 从 decisions (DB holding_strategies) 中读取 analysis 的买入区 - entry_low = d.get("entry_low", 0) - entry_high = d.get("entry_high", 0) - if not entry_low or not entry_high: - continue - - in_buy_zone = entry_low <= price <= entry_high - prev_in_buy_zone = state.get(code, {}).get("__buy_zone", None) - - # 状态变化时才触发 - if in_buy_zone and prev_in_buy_zone == False: - # 重新进入买入区 → 重评确认区间是否仍然有效 - outputs.append(f"🔄 {name}({code}) {price} → 重新进入买入区{entry_low}~{entry_high}，触发技术面重评") - do_reassess = True - elif not in_buy_zone and prev_in_buy_zone == True: - # 离开买入区 → 立即重评，更新止损/止盈/区间 - outputs.append(f"🔄 {name}({code}) {price} → 离开买入区{entry_low}~{entry_high}，立即技术面重评") - do_reassess = True - else: - do_reassess = False - - if do_reassess and HAS_REASSESS: - try: - cost = d.get("cost", 0) or 0 - shares = d.get("shares", 0) or 0 - profit_pct = (price - cost) / cost * 100 if cost else 0 - is_deep_loss = profit_pct < -20 - sentiment = "neutral" - if d.get("tech_snapshot"): - if "bearish" in d["tech_snapshot"]: - sentiment = "bearish" - elif "bullish" in d["tech_snapshot"]: - sentiment = "bullish" - - # 调用技术面驱动重评（非机械百分比） - result = reassess_strategy( - code, name, price, cost, shares, - current_action=d.get("action", ""), - volume_signal="中性", sentiment=sentiment, - ) - outputs.append(f" 📊 新策略: 损{result['stop_loss']} 盈{result['take_profit']} 区{result['entry_low']}~{result['entry_high']} RR={result['rr_ratio']}") - reassesed_codes.append(code) - except Exception as e: - outputs.append(f" ⚠️ 重评失败: {e}") - - # 更新买入区状态 - if "__buy_zone" not in state.get(code, {}): - if code not in state: - state[code] = {} - state[code]["__buy_zone"] = in_buy_zone - state_updated = True - - # 如果有重评过的股票，更新 DB holding_strategies（此前写入 decisions.json，已废弃） - if reassesed_codes and HAS_REASSESS: - # ── 5分钟冷却：regenerate_all 开销太大，不每2分钟跑一次 ── - _regen_marker = "/tmp/price_monitor_regen_at" - _skip_regen = False - try: - if os.path.exists(_regen_marker): - with open(_regen_marker) as _f: - _last_regen = float(_f.read().strip()) - if time.time() - _last_regen < 300: - _skip_regen = True - except: - pass - - if _skip_regen: - outputs.append(f" ⏭ 跳过全量重评（距上次<5min），下次再跑") - else: - try: - from strategy_lifecycle import regenerate_all - r = regenerate_all(stdout=False) - outputs.append(f" ✅ 策略已全量重评: {r.get('ok',0)}/{r.get('total',0)}成功") - outputs.append(f" 📌 触发股票: {', '.join(reassesed_codes)}") - try: - with open(_regen_marker, "w") as _f: - _f.write(str(time.time())) - except: - pass - except Exception as e: - outputs.append(f" ⚠️ 全量重评失败: {e}") - - # === 第四步：输出 === - now_str = datetime.now().strftime("%H:%M:%S") - elapsed = time.time() - start - - if outputs: - print(f"\n🔔 {now_str}{label}") - for o in outputs: - print(o) - print(f"\n{json.dumps({'type':'价格监控','time':now_str,'triggers':outputs}, ensure_ascii=False)}") - else: - # 无触发时 SILENT（中继不推送） - print(f"[SILENT]{label} 价格正常 | {refreshed}只已刷新 | {elapsed:.1f}s") - - if state_updated: - save_state(state) - - # 输出耗时 - print(f"⏱{label} {elapsed:.1f}s", flush=True) - - # 清理进程锁 - try: - os.remove("/tmp/price_monitor.lock") - except Exception: - pass - - -def main(): - """每cron触发跑一轮""" - run_once() - - -if __name__ == "__main__": - main() +#!/usr/bin/env python3 +"""price_monitor.py — 高频价格监控脚本（批量版） +规则：进入区间报一次，离开区间报一次，中间不重复。 +每次运行时一次性刷新所有持仓+自选股的实时价。 +""" +import urllib.request +import os, sys, time, json +import sqlite3 +from datetime import datetime + +from mo_data import read_decisions + +BREACH_PATH = "/home/hmo/.hermes/zone_breach.json" +STATE_PATH = "/home/hmo/.hermes/price_trigger_state.json" +EVENTS_PATH = "/home/hmo/web-dashboard/data/price_events.json" + +# DB 模块（同步实时价到 mofin.db） +sys.path.insert(0, "/home/hmo/MoFin") +try: + from mofin_db import get_conn, DB_PATH + from mo_models import calc_total_mv, calc_total_assets + HAS_DB = True +except ImportError: + HAS_DB = False + +# 策略重评依赖（技术面驱动，非机械百分比） +sys.path.insert(0, "/home/hmo/web-dashboard") +try: + from strategy_lifecycle import reassess_strategy, reassess_with_context + HAS_REASSESS = True +except ImportError: + HAS_REASSESS = False + +UA = "Mozilla/5.0" + +# ── XMPP推送 ────────────────────────────────────────────────────────── +XMPP_USER = "hmo@yoin.fun" +XMPP_BRIDGE = "http://127.0.0.1:5805/" + +def push_to_xmpp(text): + """通过知微 HTTP bridge 推送到Dad私信""" + if not text.strip(): + return + try: + payload = json.dumps({ + "to": XMPP_USER, + "body": text.strip(), + "type": "chat", + }).encode("utf-8") + req = urllib.request.Request(XMPP_BRIDGE, data=payload, headers={"Content-Type": "application/json"}) + urllib.request.urlopen(req, timeout=5) + except Exception as e: + print(f"[XMPP推送失败] {e}", file=sys.stderr) + +# ── 批量拉取价格 ────────────────────────────────────────────────────────── + +def fetch_all_prices(codes): + """腾讯批量行情API：一次请求拉取所有股票（A股+港股） + A股：sh600110 / sz000001 + 港股：hk00700 + 返回 {code: (price, change, change_pct)} + """ + if not codes: + return {} + + # 构建批量查询串 + symbols = [] + code_map = {} # symbol -> original_code + for code in codes: + code_s = str(code).strip() + if len(code_s) == 6: + # A股：沪市以5/6/9开头，深市以0/3开头 + if code_s.startswith(('5', '6', '9')): + sym = f"sh{code_s}" + else: + sym = f"sz{code_s}" + else: + sym = f"hk{code_s}" + symbols.append(sym) + code_map[sym] = code_s + + url = f"http://qt.gtimg.cn/q={','.join(symbols)}" + try: + req = urllib.request.Request(url, headers={"User-Agent": UA}) + with urllib.request.urlopen(req, timeout=10) as r: + text = r.read().decode("gbk") + except Exception as e: + print(f"⚠️ 批量拉取失败: {e}", file=sys.stderr) + return {} + + results = {} + for line in text.strip().split("\n"): + line = line.strip() + if not line or "=" not in line: + continue + try: + # 格式: v_sh600110="1~诺德股份~600110~11.84~11.90~..." + raw_value = line.split("=", 1)[1].strip().strip('"').strip(";") + fields = raw_value.split("~") + if len(fields) < 6: + continue + sym = line.split("=", 1)[0].strip().lstrip("v_") + orig_code = code_map.get(sym) + if not orig_code: + continue + price = float(fields[3]) if fields[3] else 0 + prev_close = float(fields[4]) if fields[4] else 0 + change = price - prev_close if prev_close > 0 else 0 + change_pct = fields[32] if len(fields) > 32 and fields[32] else "0" + results[orig_code] = (price, change, change_pct) + except (ValueError, IndexError): + continue + + return results + + +def refresh_data_prices(): + """一次性刷新所有持仓+自选股的实时价（完全DB版，不写JSON）""" + all_codes = set() + + # 从DB读所有需要拉取价格的代码 + try: + conn = get_conn() + for r in conn.execute("SELECT code FROM holdings WHERE is_active=1"): + all_codes.add(r['code']) + for r in conn.execute("SELECT code FROM watchlist_stocks"): + all_codes.add(r['code']) + for r in conn.execute("SELECT code FROM holding_strategies WHERE status='active'"): + all_codes.add(r['code']) + conn.close() + except Exception as e: + print(f"⚠️ 从DB读代码失败: {e}", file=sys.stderr) + return 0 + + if not all_codes: + return 0 + + # 一次性批量拉取 + prices = fetch_all_prices(list(all_codes)) + updated = len(prices) + + # === 弹性同步实时价到 mofin.db === + # 防死锁策略（经2026-07-14 WAL死锁复盘改进）： + # ① 启动时 checkpoint WAL（清理残留事务） + # ② 统一 BEGIN IMMEDIATE 包裹整个写操作 + # ③ 5次重试 + 指数退避: 1s → 2s → 4s → 8s → 16s（共~31s） + # ④ get_conn() 的 busy_timeout=30000 保证等待上限 + # ⑤ 每个写操作检查返回值，任一失败立即 rollback + 重试 + # ⑥ try/finally 确保连接始终释放 + if HAS_DB and prices: + # 先checkpoint一次，清理上次被kill残留的WAL + try: + c = get_conn() + c.execute("PRAGMA wal_checkpoint(TRUNCATE)") + c.close() + except Exception: + pass + + max_tries = 5 + conn = None + for db_attempt in range(max_tries): + try: + conn = get_conn() + # BEGIN IMMEDIATE 立即获取写锁——失败则等 busy_timeout(30s) + conn.execute("BEGIN IMMEDIATE") + + # ── 构建 holdings 更新数据 ── + db_holdings = [] + for r in conn.execute("SELECT * FROM holdings WHERE is_active=1"): + h = dict(r) + code = str(h.get('code', '')) + if code in prices: + price_val, _, change_pct = prices[code] + if price_val > 0: + h['price'] = round(price_val, 2) + h['change_pct'] = float(change_pct) if change_pct else 0 + db_holdings.append(h) + + # ── 写 holdings 表 ── + for h in db_holdings: + currency = str(h.get('currency', 'CNY')).upper() + if currency not in ('CNY', 'HKD'): + raise ValueError(f"非法币种: {currency}") + conn.execute(""" + INSERT INTO holdings (code, name, shares, cost, price, market_value, + change_pct, currency, position_pct, added_at, is_active) + VALUES (?,?,?,?,?,?,?,?,?,datetime('now','localtime'),1) + ON CONFLICT(code) DO UPDATE SET + name=excluded.name, shares=excluded.shares, cost=excluded.cost, + price=excluded.price, market_value=excluded.market_value, + change_pct=excluded.change_pct, currency=excluded.currency, + position_pct=excluded.position_pct + """, ( + h.get('code'), h.get('name'), h.get('shares', 0), + h.get('cost'), h.get('price'), + h.get('market_value'), h.get('change_pct'), + h.get('currency', 'CNY'), h.get('position_pct'), + )) + + # ── 写 portfolio_summary ── + mv = calc_total_mv(db_holdings) + existing = conn.execute( + 'SELECT cash, frozen_cash FROM portfolio_summary WHERE id=1' + ).fetchone() + db_cash = existing['cash'] if existing else 0.0 + db_frozen = existing['frozen_cash'] if existing else 0.0 + assets = calc_total_assets({'holdings': db_holdings, 'cash': db_cash, 'frozen_cash': db_frozen}) + position_pct = round(mv / assets * 100, 2) if assets > 0 else 0 + conn.execute(""" + INSERT INTO portfolio_summary (id, total_assets, total_mv, stock_value, + cash, frozen_cash, position_pct, total_pnl, currency, updated_at) + VALUES (1,?,?,?,?,?,?,?,?,datetime('now','localtime')) + ON CONFLICT(id) DO UPDATE SET + total_assets=excluded.total_assets, total_mv=excluded.total_mv, + stock_value=excluded.stock_value, cash=excluded.cash, + frozen_cash=excluded.frozen_cash, position_pct=excluded.position_pct, + total_pnl=excluded.total_pnl, currency=excluded.currency, + updated_at=datetime('now','localtime') + """, ( + assets, mv, mv, db_cash, db_frozen, + position_pct, 0, 'CNY', + )) + + # ── 写 live_prices ── + for h in db_holdings: + code = h.get('code', '') + if code: + p = h.get('price', 0) + cp = h.get('change_pct', 0) + conn.execute( + "INSERT OR REPLACE INTO live_prices (code, price, change_pct, updated_at) " + "VALUES (?,?,?,datetime('now','localtime'))", + (code, p, cp) + ) + # 补充策略股/自选股的价格（不在holdings中的） + for code, pdata in prices.items(): + if code not in {h.get('code') for h in db_holdings}: + price_val = pdata[0] if isinstance(pdata, (list, tuple)) else pdata.get('price', 0) + cp_val = pdata[1] if isinstance(pdata, (list, tuple)) else pdata.get('change_pct', 0) + conn.execute( + "INSERT OR REPLACE INTO live_prices (code, price, change_pct, updated_at) " + "VALUES (?,?,?,datetime('now','localtime'))", + (code, price_val, cp_val) + ) + + conn.commit() + conn.close() + conn = None + if db_attempt > 0: + print(f"DB同步成功（第{db_attempt+1}次重试）") + break # success + + except (sqlite3.OperationalError, sqlite3.DatabaseError) as e: + if conn: + try: conn.rollback() + except Exception: pass + try: conn.close() + except Exception: pass + conn = None + err_str = str(e) + if "locked" in err_str or "cannot commit" in err_str or "busy" in err_str: + if db_attempt < max_tries - 1: + wait = 2 ** db_attempt # 1, 2, 4, 8, 16 + print(f"⏳ DB锁(尝试{db_attempt+1}/{max_tries}): {e} → {wait}s后重试", file=sys.stderr) + time.sleep(wait) + else: + print(f"❌ DB锁(重试{max_tries}次耗尽): {e}", file=sys.stderr) + else: + print(f"❌ DB错误: {e}", file=sys.stderr) + break + except Exception as e: + if conn: + try: conn.rollback() + except Exception: pass + try: conn.close() + except Exception: pass + conn = None + print(f"⚠️ DB同步异常: {e}", file=sys.stderr) + break + else: + # for-else: loop exhausted without break + print("❌ DB同步失败（所有重试耗尽）", file=sys.stderr) + # 尝试紧急 WAL checkpoint（释放死锁） + try: + c = sqlite3.connect(str(DB_PATH), timeout=1) + c.execute("PRAGMA wal_checkpoint(TRUNCATE)") + c.close() + print(" ↪ 紧急WAL checkpoint完成", file=sys.stderr) + except Exception as we: + print(f" ↪ WAL checkpoint也失败: {we}", file=sys.stderr) + + return updated + + +# ── 区间偏离检测 ────────────────────────────────────────────────────────── + +def load_state(): + try: + with open(STATE_PATH) as f: + return json.load(f) + except: + return {} + +def save_state(state): + os.makedirs(os.path.dirname(STATE_PATH), exist_ok=True) + with open(STATE_PATH, 'w') as f: + json.dump(state, f, ensure_ascii=False, indent=2) + +def load_breaches(): + try: + with open(BREACH_PATH) as f: + return json.load(f) + except: + return {} + +def save_breaches(data): + os.makedirs(os.path.dirname(BREACH_PATH), exist_ok=True) + with open(BREACH_PATH, 'w') as f: + json.dump(data, f, ensure_ascii=False, indent=2) + + +def load_events(): + try: + with open(EVENTS_PATH) as f: + return json.load(f) + except: + return {"events": []} + + +def save_events(events): + os.makedirs(os.path.dirname(EVENTS_PATH), exist_ok=True) + with open(EVENTS_PATH, 'w') as f: + json.dump(events, f, ensure_ascii=False, indent=2) + + +def record_event(code, name, event_type, price, trigger_value, event_label=""): + """记录一次价格触发事件 — 双写：DB price_events 表（权威）+ price_events.json（遗留读取方兼容）""" + now = datetime.now().isoformat() + date_str = datetime.now().strftime("%Y-%m-%d") + + # 1. 写 DB 表（权威存储）。price_events.code 有 FK -> stocks(code)， + # 未注册的股票（新候选/港股）会先注册再写事件，否则 FK 失败事件丢失。 + if HAS_DB: + try: + from mofin_db import get_conn, write_price_event + _c = get_conn() + _exch, _typ = ("HK", "H") if len(str(code)) == 5 else (("SH", "A") if str(code).startswith(("6", "9")) else ("SZ", "A")) + _c.execute("INSERT OR IGNORE INTO stocks (code, name, exchange, type, updated_at) VALUES (?,?,?,?,?)", + (str(code), name or str(code), _exch, _typ, now)) + _c.commit() + write_price_event(_c, code=code, name=name, event_type=event_type, + price=round(price, 2), trigger_value=str(trigger_value), + event_label=event_label) + _c.close() + except Exception as e: + print(f"[price_events DB写入失败] {e}", file=sys.stderr) + + # 2. 写 JSON（遗留读取方：mo_config/strategy_feedback/system_health_check 还在读） + events = load_events() + events["events"].append({ + "code": code, + "name": name, + "event_type": event_type, # entry_zone, stop_loss, take_profit, exit_zone + "price": round(price, 2), + "trigger_value": trigger_value, + "event_label": event_label, + "timestamp": now, + "date": date_str, + }) + # 保留最近10000条 + events["events"] = events["events"][-10000:] + save_events(events) + + +def get_trigger_zones(trigger): + """返回该trigger所有可监控的区间列表，跳过已执行的batch""" + zones = [] + for key, label in [ + ("entry_zone", "加仓区间"), + ("batch1_price", "试仓区间"), + ("batch2_price", "加仓区间"), + ("take_profit_zone", "止盈区间"), + ("watch_low", "关注区间"), + ("watch_high", "减仓区间"), + ("watch_break", "止损区间") + ]: + status_key = key.replace("_price", "_status") + if status_key in trigger and trigger[status_key] == "executed": + continue + val = trigger.get(key, "") + if val and "~" in val: + try: + parts = val.split("~") + lo, hi = float(parts[0]), float(parts[1]) + zones.append((key, label, lo, hi)) + except: + pass + sl = trigger.get("stop_loss", "") + if sl: + try: + sl_price = float(sl) if isinstance(sl, (int, float)) else float(sl) + zones.append(("stop_loss", "止损", 0, sl_price)) + except: + pass + return zones + + +def _cleanup_lock(): + """清理进程锁文件""" + try: + os.remove("/tmp/price_monitor.lock") + except Exception: + pass + +def _handle_sigterm(signum, frame): + """收到SIGTERM时清理锁文件后退出""" + _cleanup_lock() + sys.exit(0) + +def run_once(round_label=""): + """执行一轮完整的监控流程""" + import os, signal # 必须在开头import，否则os变量会被后面的局部import绑定覆盖 + signal.signal(signal.SIGTERM, _handle_sigterm) + os.nice(10) # 降低优先级，避免与DB其他写操作抢占 + # ── 进程锁：同一时间只跑一个实例 ── + _lk = "/tmp/price_monitor.lock" + _pid = None + try: + with open(_lk) as _f: + _pid = int(_f.read().strip()) + os.kill(_pid, 0) + print(f"[LOCK] 已有实例(PID {_pid})在运行，跳过本轮", file=sys.stderr, flush=True) + return + except (FileNotFoundError, ProcessLookupError, ValueError): + pass + with open(_lk, "w") as _f: + _f.write(str(os.getpid())) + + label = f" [{round_label}]" if round_label else "" + start = time.time() + TIME_BUDGET = 90 # 预留30s给输出和清理，90s内必须完成核心逻辑 + + # === 第一步：一次性刷新所有价格 === + refreshed = refresh_data_prices() + + # === 第二步：检查触发条件 === + try: + dec = read_decisions() + except: + print(f"❌{label} 无法读取decisions(DB)", file=sys.stderr) + return + + active = [d for d in dec.get("decisions", []) if d.get("status") == "active"] + state = load_state() + outputs = [] + state_updated = False + # 时间冷却：同股同区间30分钟内不重复推 + _push_cooldown = {} + _cooldown_file = "/home/hmo/.hermes/.price_push_cooldown.json" + try: + import os + if os.path.exists(_cooldown_file): + with open(_cooldown_file) as _f: + _push_cooldown = json.load(_f) + except Exception: + _push_cooldown = {} + + def _can_push(code, zone_key): + now = time.time() + key = f"{code}_{zone_key}" + last = _push_cooldown.get(key, 0) + if now - last < 1800: # 30分钟 + return False + _push_cooldown[key] = now + # 持久化写入 + try: + with open(_cooldown_file, "w") as _f: + json.dump(_push_cooldown, _f) + except Exception: + pass + return True + + # 收集所有需要检查的代码 + check_codes = set() + for d in active: + trig = d.get("trigger", {}) + if trig: + check_codes.add(d["code"]) + + # 批量拉取这些股票的价格 + prices = fetch_all_prices(list(check_codes)) + + for d in active: + code = d["code"] + trig = d.get("trigger", {}) + if not trig: + continue + + zones = get_trigger_zones(trig) + if not zones: + continue + + price_info = prices.get(code) + if not price_info: + continue + price, _, _ = price_info + if price == 0: + continue + + name = d.get("name", code) + if code not in state: + state[code] = {} + + # 时间预算检查：如果超时，跳过重评只做状态记录 + _budget_low = (time.time() - start) > TIME_BUDGET + + for key, label, lo, hi in zones: + in_zone = lo <= price <= hi + prev_in_zone = state[code].get(key, None) + + if in_zone and prev_in_zone != True: + if key == "stop_loss": + outputs.append(f"⚠️ {name}({code}) {price} → 跌破止损{hi}！") + record_event(code, name, "stop_loss", price, str(hi)) + # 止损触发 → 立即重评并推送给Dad（时间不够则直接推原始告警） + if _budget_low: + outputs.append(f" 📨 止损触发(超时跳过重评)→已推送Dad") + if _can_push(code, "stop_loss"): + push_to_xmpp(f"⚠️ {name}({code}) {price} → 跌破止损{hi}！") + else: + try: + cost = d.get("cost", 0) or 0 + shares = d.get("shares", 0) or 0 + current_action = d.get("action", "") + result = reassess_with_context(code, name, price, cost, shares, current_action) + if result: + timing_signal = result.get("timing_signal", "") + action = result.get("action", "") + if "买入" in timing_signal or "加仓" in timing_signal or timing_signal in ("卖出","止盈"): + buy_lo = d.get("entry_low", 0) + buy_hi = d.get("entry_high", 0) + rr = result.get("rr_ratio", 0) + if _can_push(code, "stop_loss"): + msg = f"🔔 {name}({code}) 价{price}→触发操作区间{max(buy_lo,0):.2f}~{buy_hi:.2f}，已触发重评|RR={rr}" + push_to_xmpp(msg) + outputs.append(f" 📨 止损重评→已推送Dad: {action}") + except Exception as e: + outputs.append(f" ⚠️ 止损重评失败: {e}") + else: + extra = "" + if "_price" in key: + batch_shares = trig.get(key.replace("_price", "_shares"), "") + action = trig.get(key.replace("_price", "_action"), "") + if batch_shares: + extra = f" {action}{batch_shares}股" if action else f" {batch_shares}股" + elif key in ("take_profit_zone",): + act = trig.get("take_profit_action", "") + if act: + extra = f"（{act}）" + outputs.append(f"⚡ {name}({code}) {price} → 进入{label}{lo}~{hi}{extra}") + record_event(code, name, "entry_zone", price, f"{lo}~{hi}", label) + # 进入区间 → 立即重评并推送给Dad（时间不够则跳过重评直接推原始告警） + if _budget_low: + if _can_push(code, key): + push_to_xmpp(f"⚡ {name}({code}) {price} → 进入{label}{lo}~{hi}") + outputs.append(f" 📨 区间触发(超时跳过重评)→已推送Dad") + else: + try: + cost = d.get("cost", 0) or 0 + shares = d.get("shares", 0) or 0 + current_action = d.get("action", "") + result = reassess_with_context(code, name, price, cost, shares, current_action) + if result: + timing_signal = result.get("timing_signal", "") + action = result.get("action", "") + # 格式化区间描述（止盈区lo=0时美化显示） + if key == "take_profit_zone" and lo == 0: + zone_desc = f"止盈监控(目标{hi:.0f})" + else: + zone_desc = f"操作区间{lo}~{hi}" + if "买入" in timing_signal or "加仓" in timing_signal or timing_signal in ("卖出","止盈"): + rr = result.get("rr_ratio", 0) + if _can_push(code, key): + msg = f"🔔 {name}({code}) 价{price}→触发{zone_desc}，已触发重评|RR={rr}" + push_to_xmpp(msg) + outputs.append(f" 📨 区间触发重评→已推送Dad: {action}") + else: + reason = f"重评结果:{timing_signal}，不构成操作建议" + outputs.append(f" 📋 本地日志(不推): {reason}") + except Exception as e: + outputs.append(f" ⚠️ 区间重评失败: {e}") + state[code][key] = True + state_updated = True + + elif not in_zone and prev_in_zone == True: + if key != "stop_loss": + outputs.append(f"📌 {name}({code}) {price} → 离开{label}{lo}~{hi}") + state[code][key] = False + state_updated = True + + # === 第三步：买入区偏离检测 + 自动重评 === + reassesed_codes = [] + # 先做急跌检测（仅持仓，自选股不推送暴跌告警） + holdings_codes = {d["code"] for d in active if (d.get("shares") or 0) > 0} + for d in active: + code = d["code"] + # 非持仓跳过 + if code not in holdings_codes: + continue + name = d.get("name", code) + price_info = prices.get(code) + if not price_info: + continue + price, _, change_pct = price_info + if price == 0: + continue + # 单日跌幅>7%告警（不依赖zone边界，盘中急跌即触发） + try: + cp = float(change_pct) if change_pct else 0 + except: + cp = 0 + if cp <= -7: + prev_alert = state.get(code, {}).get("__sharp_decline_triggered", False) + if not prev_alert: + stop_loss = d.get("stop_loss", 0) + sl_note = f" 止损{stop_loss}" if stop_loss else "" + msg = f"🔻 {name}({code}) {price} 暴跌{cp:.1f}%！{sl_note}" + push_to_xmpp(msg) + outputs.append(msg) + state.setdefault(code, {})["__sharp_decline_triggered"] = True + state_updated = True + # 立即持久化，防止后续超时导致状态丢失而重复推送 + save_state(state) + elif cp > -5: + # 反弹后清除告警标记，下次再跌还能报 + state.setdefault(code, {}).pop("__sharp_decline_triggered", None) + + for d in active: + code = d["code"] + name = d.get("name", code) + price_info = prices.get(code) + if not price_info: + continue + price, _, _ = price_info + if price == 0: + continue + + # 从 decisions (DB holding_strategies) 中读取 analysis 的买入区 + entry_low = d.get("entry_low", 0) + entry_high = d.get("entry_high", 0) + if not entry_low or not entry_high: + continue + + in_buy_zone = entry_low <= price <= entry_high + prev_in_buy_zone = state.get(code, {}).get("__buy_zone", None) + + # 状态变化时才触发 + if in_buy_zone and prev_in_buy_zone == False: + # 重新进入买入区 → 重评确认区间是否仍然有效 + outputs.append(f"🔄 {name}({code}) {price} → 重新进入买入区{entry_low}~{entry_high}，触发技术面重评") + do_reassess = True + elif not in_buy_zone and prev_in_buy_zone == True: + # 离开买入区 → 立即重评，更新止损/止盈/区间 + outputs.append(f"🔄 {name}({code}) {price} → 离开买入区{entry_low}~{entry_high}，立即技术面重评") + do_reassess = True + else: + do_reassess = False + + if do_reassess and HAS_REASSESS: + try: + cost = d.get("cost", 0) or 0 + shares = d.get("shares", 0) or 0 + profit_pct = (price - cost) / cost * 100 if cost else 0 + is_deep_loss = profit_pct < -20 + sentiment = "neutral" + if d.get("tech_snapshot"): + if "bearish" in d["tech_snapshot"]: + sentiment = "bearish" + elif "bullish" in d["tech_snapshot"]: + sentiment = "bullish" + + # 调用技术面驱动重评（非机械百分比） + result = reassess_strategy( + code, name, price, cost, shares, + current_action=d.get("action", ""), + volume_signal="中性", sentiment=sentiment, + ) + outputs.append(f" 📊 新策略: 损{result['stop_loss']} 盈{result['take_profit']} 区{result['entry_low']}~{result['entry_high']} RR={result['rr_ratio']}") + reassesed_codes.append(code) + except Exception as e: + outputs.append(f" ⚠️ 重评失败: {e}") + + # 更新买入区状态 + if "__buy_zone" not in state.get(code, {}): + if code not in state: + state[code] = {} + state[code]["__buy_zone"] = in_buy_zone + state_updated = True + + # 如果有重评过的股票，更新 DB holding_strategies（此前写入 decisions.json，已废弃） + if reassesed_codes and HAS_REASSESS: + # ── 5分钟冷却：regenerate_all 开销太大，不每2分钟跑一次 ── + _regen_marker = "/tmp/price_monitor_regen_at" + _skip_regen = False + try: + if os.path.exists(_regen_marker): + with open(_regen_marker) as _f: + _last_regen = float(_f.read().strip()) + if time.time() - _last_regen < 300: + _skip_regen = True + except: + pass + + if _skip_regen: + outputs.append(f" ⏭ 跳过全量重评（距上次<5min），下次再跑") + else: + try: + from strategy_lifecycle import regenerate_all + r = regenerate_all(stdout=False) + outputs.append(f" ✅ 策略已全量重评: {r.get('ok',0)}/{r.get('total',0)}成功") + outputs.append(f" 📌 触发股票: {', '.join(reassesed_codes)}") + try: + with open(_regen_marker, "w") as _f: + _f.write(str(time.time())) + except: + pass + except Exception as e: + outputs.append(f" ⚠️ 全量重评失败: {e}") + + # === 第四步：输出 === + now_str = datetime.now().strftime("%H:%M:%S") + elapsed = time.time() - start + + if outputs: + print(f"\n🔔 {now_str}{label}") + for o in outputs: + print(o) + print(f"\n{json.dumps({'type':'价格监控','time':now_str,'triggers':outputs}, ensure_ascii=False)}") + else: + # 无触发时 SILENT（中继不推送） + print(f"[SILENT]{label} 价格正常 | {refreshed}只已刷新 | {elapsed:.1f}s") + + if state_updated: + save_state(state) + + # 输出耗时 + print(f"⏱{label} {elapsed:.1f}s", flush=True) + + # 清理进程锁 + try: + os.remove("/tmp/price_monitor.lock") + except Exception: + pass + + +def main(): + """每cron触发跑一轮""" + run_once() + + +if __name__ == "__main__": + main()