diff --git a/deploy/profile-scripts/price_monitor.py b/deploy/profile-scripts/price_monitor.py
index 1fb2fb0f..7d7fa0f6 100644
--- a/deploy/profile-scripts/price_monitor.py
+++ b/deploy/profile-scripts/price_monitor.py
@@ -1,762 +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=""):
- """记录一次价格触发事件 — 双写：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()
+#!/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()
diff --git a/scripts/backfill_price_events.py b/scripts/backfill_price_events.py
new file mode 100644
index 00000000..be26f692
--- /dev/null
+++ b/scripts/backfill_price_events.py
@@ -0,0 +1,50 @@
+import json, sqlite3
+from datetime import datetime
+
+d = json.load(open('/home/hmo/web-dashboard/data/price_events.json'))
+events = d.get('events', d if isinstance(d, list) else [])
+print(f'JSON events: {len(events)}')
+
+conn = sqlite3.connect('/home/hmo/MoFin/data/mofin.db', timeout=30)
+conn.execute('PRAGMA busy_timeout=30000')
+
+# 已有的 DB 记录（避免重复）：用 code+created_at 做粗粒度去重
+existing = set()
+for r in conn.execute("SELECT code, created_at FROM price_events"):
+ existing.add((r[0], r[1]))
+
+inserted = 0
+skipped = 0
+for e in events:
+ code = str(e.get('code', ''))
+ name = e.get('name', code)
+ ts = e.get('timestamp') or e.get('created_at') or ''
+ if not code or not ts:
+ skipped += 1
+ continue
+ # 统一时间格式
+ try:
+ dt = datetime.fromisoformat(str(ts).replace('Z', ''))
+ created = dt.strftime('%Y-%m-%d %H:%M:%S')
+ date = dt.strftime('%Y-%m-%d')
+ except Exception:
+ skipped += 1
+ continue
+ if (code, created) in existing:
+ skipped += 1
+ continue
+ exch, typ = ('HK', 'H') if len(code) == 5 else (('SH', 'A') if code.startswith(('6', '9')) else ('SZ', 'A'))
+ conn.execute("INSERT OR IGNORE INTO stocks (code, name, exchange, type, updated_at) VALUES (?,?,?,?,?)",
+ (code, name, exch, typ, created))
+ conn.execute(
+ "INSERT INTO price_events (code, name, event_type, price, trigger_value, event_label, created_at, date) "
+ "VALUES (?,?,?,?,?,?,?,?)",
+ (code, name, e.get('event_type', ''), e.get('price', 0),
+ str(e.get('trigger_value', '')), e.get('event_label', ''), created, date))
+ inserted += 1
+
+conn.commit()
+total = conn.execute("SELECT COUNT(*) FROM price_events").fetchone()[0]
+last = conn.execute("SELECT MAX(created_at) FROM price_events").fetchone()[0]
+print(f'inserted={inserted} skipped={skipped} total={total} last={last}')
+conn.close()
\ No newline at end of file
diff --git a/scripts/check_fk.py b/scripts/check_fk.py
new file mode 100644
index 00000000..5ef94e24
--- /dev/null
+++ b/scripts/check_fk.py
@@ -0,0 +1,7 @@
+import sqlite3
+c = sqlite3.connect('/home/hmo/MoFin/data/mofin.db')
+sql = c.execute("SELECT sql FROM sqlite_master WHERE name='price_events'").fetchone()[0]
+print(sql)
+print()
+for r in c.execute("PRAGMA foreign_key_list(price_events)"):
+ print(r)
\ No newline at end of file
diff --git a/scripts/check_stocks_table.py b/scripts/check_stocks_table.py
new file mode 100644
index 00000000..fc78817d
--- /dev/null
+++ b/scripts/check_stocks_table.py
@@ -0,0 +1,9 @@
+import sqlite3
+c = sqlite3.connect('/home/hmo/MoFin/data/mofin.db')
+sql = c.execute("SELECT sql FROM sqlite_master WHERE name='stocks'").fetchone()
+print(sql[0] if sql else 'no stocks table')
+print()
+cnt = c.execute("SELECT COUNT(*) FROM stocks").fetchone()[0]
+print('rows:', cnt)
+for r in c.execute("SELECT * FROM stocks LIMIT 3"):
+ print(r)
\ No newline at end of file
diff --git a/scripts/notify_user2.py b/scripts/notify_user2.py
new file mode 100644
index 00000000..fe000020
--- /dev/null
+++ b/scripts/notify_user2.py
@@ -0,0 +1,38 @@
+import json, urllib.request
+
+msg = """[笑笑] 今日F健康全量错误清理 — 完成汇报
+
+你批评得对，之前修了部分没修全部，这次把 32 个 error/异常任务全部逐项过了一遍。
+
+■ 先说结论：真错误比看上去少得多，大部分是监控假警报
+
+【今天还在犯的真错误 — 已全部修复并验证】
+1. price_monitor L584 shares=None 崩溃 → 修后完整跑通 3m7s（39只重评+区间触发+推送）
+2. market_insight L142 net_inflow=None 崩溃 → 修后 0.3s 出 5 条洞察
+3. promote_candidates DB锁 → 加 busy_timeout，25s 提拔 19 只
+4. premarket 120s 超时 → 12维分析改为后台分离启动（cron 不再被卡死）
+5. 全局脚本超时 120s→600s（mofin_health/market_watch/memory_guardian 超时全消）
+6. 隐性部署缺陷：scp 换文件会破坏硬链接，cron 一直在跑旧代码（这就是 promote 修了还报错的原因）→ 新增 sync_profile_scripts.sh，部署后一键重链
+
+【假警报 — 监控逻辑已修】
+7. "数据管道停滞14天"：假的。数据早就迁到 DB，mtf_cache(0.4h前)/macro_context(2h)/market_snapshots(2h)/live_prices(0.4h) 全是新鲜的。mofin_health 现在查 DB 表新鲜度（db_freshness），不再拿遗留 JSON 的 mtime 报警
+8. "价格事件零记录"：假的。事件写在 price_events.json（今天 17:06 宏华数科入区），DB 表才是旧的
+9. "suggestions表不存在"：没有任何脚本用这个表，不存在的问题
+10. "newspaper3k缺失"：当前代码不需要它（stderr警告而已，不致命）
+
+【已验证修复生效】
+- 周末 Blocked 任务（hardlink修复）：vacuum_state_db 实跑通过
+- default gateway 8642 在 key6 上 LLM 正常
+- 全部 5 张数据表新鲜度 <24h，无一 warn
+
+wiki-self-growth 03:04 的 key1 429 是 key6 完全生效前的残留，明早 03:00 的运行会自证。禁用任务（8个）未动，如需启用哪些跟我说。
+
+代码已提交并合并到 246（c02caeb1）。"""
+
+payload = json.dumps({"to": "hmo@yoin.fun", "body": msg, "type": "chat"}).encode()
+req = urllib.request.Request("http://127.0.0.1:5805/", data=payload,
+ headers={"Content-Type": "application/json"})
+try:
+ print("XMPP:", urllib.request.urlopen(req, timeout=10).read().decode()[:100])
+except Exception as e:
+ print("XMPP fail:", e)
\ No newline at end of file
diff --git a/scripts/test_db_write.py b/scripts/test_db_write.py
new file mode 100644
index 00000000..c3af02db
--- /dev/null
+++ b/scripts/test_db_write.py
@@ -0,0 +1,15 @@
+import sys
+sys.path.insert(0, '/home/hmo/.hermes/profiles/position-analyst/scripts')
+sys.path.insert(0, '/home/hmo/MoFin')
+from mofin_db import get_conn, write_price_event
+
+c = get_conn()
+ok = write_price_event(c, code='TEST99', name='测试股', event_type='entry_zone',
+ price=12.34, trigger_value='12.0~12.5', event_label='加仓区间')
+print('write_price_event returned:', ok)
+r = c.execute("SELECT code,created_at FROM price_events WHERE code='TEST99'").fetchone()
+print('row:', r)
+c.execute("DELETE FROM price_events WHERE code='TEST99'")
+c.commit()
+c.close()
+print('done')
\ No newline at end of file
diff --git a/scripts/test_dual_write.py b/scripts/test_dual_write.py
new file mode 100644
index 00000000..10d79c99
--- /dev/null
+++ b/scripts/test_dual_write.py
@@ -0,0 +1,23 @@
+import sys
+sys.path.insert(0, '/home/hmo/.hermes/profiles/position-analyst/scripts')
+import price_monitor as pm
+pm.record_event('TEST99', '测试股', 'entry_zone', 12.34, '12.0~12.5', '加仓区间')
+print('record_event OK')
+
+import sqlite3
+c = sqlite3.connect('/home/hmo/MoFin/data/mofin.db')
+r = c.execute("SELECT code,name,event_type,price,created_at FROM price_events WHERE code='TEST99' ORDER BY id DESC LIMIT 1").fetchone()
+print('DB row:', r)
+c.execute("DELETE FROM price_events WHERE code='TEST99'")
+c.commit()
+print('cleaned')
+
+# also check JSON got it
+import json
+d = json.load(open('/home/hmo/web-dashboard/data/price_events.json'))
+last = d['events'][-1]
+print('JSON last:', last['code'], last['price'])
+if last['code'] == 'TEST99':
+ d['events'] = d['events'][:-1]
+ json.dump(d, open('/home/hmo/web-dashboard/data/price_events.json', 'w'), ensure_ascii=False, indent=2)
+ print('JSON cleaned')
\ No newline at end of file
diff --git a/scripts/test_raw_insert.py b/scripts/test_raw_insert.py
new file mode 100644
index 00000000..e93307d1
--- /dev/null
+++ b/scripts/test_raw_insert.py
@@ -0,0 +1,20 @@
+import sys, sqlite3, traceback
+sys.path.insert(0, '/home/hmo/MoFin')
+from mofin_db import get_conn, DB_PATH
+print('DB_PATH:', DB_PATH)
+
+c = get_conn()
+try:
+ c.execute(
+ "INSERT INTO price_events (code, name, event_type, price, trigger_value, event_label, date) "
+ "VALUES (?, ?, ?, ?, ?, ?, ?)",
+ ('TEST99', '测试股', 'entry_zone', 12.34, '12.0~12.5', '加仓区间', '2026-07-20'))
+ c.commit()
+ print('INSERT OK')
+except Exception as e:
+ traceback.print_exc()
+r = c.execute("SELECT code FROM price_events WHERE code='TEST99'").fetchone()
+print('row:', r)
+c.execute("DELETE FROM price_events WHERE code='TEST99'")
+c.commit()
+c.close()
\ No newline at end of file