diff --git a/deploy/profile-scripts/batch_reassess.py b/deploy/profile-scripts/batch_reassess.py index f707676e..12c598f6 100644 --- a/deploy/profile-scripts/batch_reassess.py +++ b/deploy/profile-scripts/batch_reassess.py @@ -123,30 +123,26 @@ def collect_data(code): except Exception: data["strategy_def"] = None # ── 技术指标收集(策略输入数据)── + # 从 stock_indicators 读取实时技术指标(由 price_monitor 联动更新) try: - import technical_analysis as _ta - _tech = _ta.full_analysis(code) - if _tech: - _sr = _tech.get("support_resistance", {}) - data["ta_strong_support"] = _sr.get("strong_support", 0) - data["ta_weak_support"] = _sr.get("weak_support", 0) - data["ta_pivot"] = _sr.get("pivot", 0) - data["ta_weak_resist"] = _sr.get("weak_resist", 0) - data["ta_strong_resist"] = _sr.get("strong_resist", 0) - _cs = _tech.get("candlestick", {}) - data["ta_candle"] = _cs.get("pattern", "") + "/" + _cs.get("sentiment", "") - _vol = _tech.get("volume", {}) - data["ta_volume"] = _vol.get("description", "") - import re as _re - _snap = data.get("tech_snapshot", "") - _ma = _re.search(r'MA5=([\d.]+).*?MA10=([\d.]+).*?MA20=([\d.]+).*?MA60=([\d.]+)', _snap) - if _ma: - data["ta_ma5"] = float(_ma.group(1)) - data["ta_ma10"] = float(_ma.group(2)) - data["ta_ma20"] = float(_ma.group(3)) - data["ta_ma60"] = float(_ma.group(4)) - if data["ta_ma20"] > 0: - data["ta_dist_ma20"] = round((data["price"] - data["ta_ma20"]) / data["ta_ma20"] * 100, 2) + _si = _db2.execute( + "SELECT ma5, ma10, ma20, ma60, rsi, bias60, dist_ma20, strong_support, weak_support, pivot, weak_resist, strong_resist, candle_pattern, vol_ratio " + "FROM stock_indicators WHERE code=? ORDER BY date DESC LIMIT 1", + (code,)).fetchone() + if _si: + data["ta_ma5"] = _si[0] if _si[0] else None + data["ta_ma10"] = _si[1] if _si[1] else None + data["ta_ma20"] = _si[2] if _si[2] else None + data["ta_ma60"] = _si[3] if _si[3] else None + data["ta_rsi"] = round(_si[4], 1) if _si[4] else None + data["ta_dist_ma20"] = _si[6] if _si[6] else None + data["ta_strong_support"] = _si[7] if _si[7] else None + data["ta_weak_support"] = _si[8] if _si[8] else None + data["ta_pivot"] = _si[9] if _si[9] else None + data["ta_weak_resist"] = _si[10] if _si[10] else None + data["ta_strong_resist"] = _si[11] if _si[11] else None + data["ta_candle"] = _si[12] or "" + data["ta_volume"] = "" except Exception: pass try: diff --git a/deploy/profile-scripts/price_monitor.py b/deploy/profile-scripts/price_monitor.py index 0379069e..84dbe93c 100644 --- a/deploy/profile-scripts/price_monitor.py +++ b/deploy/profile-scripts/price_monitor.py @@ -376,6 +376,17 @@ def refresh_data_prices(): "VALUES (?,?,?,datetime('now','localtime'))", (code, p, cp) ) + + # 实时技术指标联动更新 + try: + from realtime_indicators import update_single + for h in db_holdings: + _ri_code = h.get("code", "") + _ri_price = h.get("price", 0) + if _ri_code and _ri_price: + update_single(_ri_code, float(_ri_price)) + except Exception as e: + print(" realtime_indicators error:", e, file=sys.stderr) # 补充策略股/自选股的价格(不在holdings中的) for code, pdata in prices.items(): if code not in {h.get('code') for h in db_holdings}: diff --git a/deploy/profile-scripts/realtime_indicators.py b/deploy/profile-scripts/realtime_indicators.py new file mode 100644 index 00000000..53abd37e --- /dev/null +++ b/deploy/profile-scripts/realtime_indicators.py @@ -0,0 +1,165 @@ +#!/usr/bin/env python3 +"""realtime_indicators.py — 盘中实时技术指标计算+写入stock_indicators + +触发:price_monitor 更新价格后联动调用 +输出:MA/支撑阻力/RSI/bias60/dist_ma20/candle_pattern 写入 stock_indicators +""" +import os, sys, sqlite3 +from datetime import datetime + +DB = "/home/hmo/MoFin/data/mofin.db" + + +def calc_realtime_indicators(code, price, date_str=None): + """计算单只股票的实时技术指标""" + if not date_str: + date_str = datetime.now().strftime("%Y-%m-%d") + + conn = sqlite3.connect(DB, timeout=30) + + # 读最近60日K线(计算MA/RSI/bias60需要) + rows = conn.execute( + "SELECT date, close FROM stock_daily WHERE code=? ORDER BY date DESC LIMIT 60", + (code,)).fetchall() + + if not rows or len(rows) < 5: + conn.close() + return None + + closes = [r[1] for r in rows if r[1]] + if not closes: + conn.close() + return None + + result = {"code": code, "date": date_str, "updated_at": datetime.now().isoformat()} + + # MA(用最近N日收盘价,当天价格替代最新收盘价) + closes_with_today = [price] + closes # 最新价在前 + result["ma5"] = round(sum(closes_with_today[:5]) / min(5, len(closes_with_today)), 2) if closes_with_today else None + result["ma10"] = round(sum(closes_with_today[:10]) / min(10, len(closes_with_today)), 2) if closes_with_today else None + result["ma20"] = round(sum(closes_with_today[:20]) / min(20, len(closes_with_today)), 2) if closes_with_today else None + result["ma60"] = round(sum(closes_with_today[:60]) / min(60, len(closes_with_today)), 2) if closes_with_today else None + + # RSI(14日) + if len(closes_with_today) >= 15: + gains = [] + losses = [] + for i in range(1, min(15, len(closes_with_today))): + diff = closes_with_today[i-1] - closes_with_today[i] + if diff > 0: + gains.append(diff) + losses.append(0) + else: + gains.append(0) + losses.append(abs(diff)) + avg_gain = sum(gains) / len(gains) if gains else 0 + avg_loss = sum(losses) / len(losses) if losses else 0.01 + rs = avg_gain / avg_loss + result["rsi"] = round(100 - 100 / (1 + rs), 2) + + # bias60(偏离60日均线百分比) + if result.get("ma60") and result["ma60"] > 0: + result["bias60"] = round((price - result["ma60"]) / result["ma60"] * 100, 2) + + # dist_ma20(距MA20百分比) + if result.get("ma20") and result["ma20"] > 0: + result["dist_ma20"] = round((price - result["ma20"]) / result["ma20"] * 100, 2) + + # 支撑阻力位(简化版:基于近期高低点) + recent_lows = sorted(closes[:20]) # 近20日最低 + recent_highs = sorted(closes[:20], reverse=True) + result["weak_support"] = round(recent_lows[0], 2) if recent_lows else None + result["strong_support"] = round(recent_lows[2], 2) if len(recent_lows) > 2 else result["weak_support"] + result["weak_resist"] = round(recent_highs[0], 2) if recent_highs else None + result["strong_resist"] = round(recent_highs[2], 2) if len(recent_highs) > 2 else result["weak_resist"] + result["pivot"] = round((result.get("strong_support", price) + result.get("strong_resist", price)) / 2, 2) + + # ATR(14日平均真实波幅) + if len(rows) >= 15: + trs = [] + for i in range(14): + h = closes[i] if i < len(closes) else closes[-1] + l = closes[i+1] if i+1 < len(closes) else closes[-1] + c_prev = closes[i+1] if i+1 < len(closes) else closes[-1] + tr = max(h - l, abs(h - c_prev), abs(l - c_prev)) + trs.append(tr) + result["atr"] = round(sum(trs) / len(trs), 2) if trs else None + + # 量比(当日量/5日均量) + volumes = conn.execute( + "SELECT volume FROM stock_daily WHERE code=? ORDER BY date DESC LIMIT 6", + (code,)).fetchall() + if volumes and len(volumes) >= 2: + today_vol = volumes[0][0] or 0 + avg_vol = sum(v[0] for v in volumes[1:6]) / min(5, len(volumes)-1) if len(volumes) > 1 else 1 + result["vol_ratio"] = round(today_vol / avg_vol, 2) if avg_vol > 0 else None + + conn.close() + return result + + +def save_indicators(data): + """写入 stock_indicators 表(INSERT OR REPLACE)""" + if not data: + return + conn = sqlite3.connect(DB, timeout=30) + + # 检查是否已有该日期记录 + existing = conn.execute( + "SELECT id FROM stock_indicators WHERE code=? AND date=?", + (data["code"], data["date"])).fetchone() + + fields = [k for k in data if k not in ("code", "date")] + cols = ", ".join(["code", "date"] + fields) + vals = ", ".join(["?"] * (2 + len(fields))) + updates = ", ".join(f"{f}=excluded.{f}" for f in fields) + + conn.execute( + f"INSERT INTO stock_indicators ({cols}) VALUES ({vals}) " + f"ON CONFLICT(code, date) DO UPDATE SET {updates}", + [data["code"], data["date"]] + [data[f] for f in fields]) + conn.commit() + conn.close() + + +def update_all_active(): + """批量更新所有活跃股票的实时指标(每日收盘后)""" + conn = sqlite3.connect(DB, timeout=30) + today = datetime.now().strftime("%Y-%m-%d") + + codes = set() + for r in conn.execute("SELECT code FROM holdings WHERE is_active=1"): + codes.add(r[0]) + for r in conn.execute("SELECT code FROM watchlist_stocks"): + codes.add(r[0]) + for r in conn.execute("SELECT code FROM holding_strategies WHERE status='active'"): + codes.add(r[0]) + + ok = 0 + for code in codes: + # 从 stock_daily 读收盘价 + row = conn.execute( + "SELECT close FROM stock_daily WHERE code=? ORDER BY date DESC LIMIT 1", + (code,)).fetchone() + if not row or not row[0]: + continue + price = float(row[0]) + + data = calc_realtime_indicators(code, price, today) + if data: + save_indicators(data) + ok += 1 + + conn.close() + print(f"实时指标更新: {ok}/{len(codes)} 只") + + +def update_single(code, price): + """单只股票实时更新(price_monitor 联动调用)""" + data = calc_realtime_indicators(code, price) + if data: + save_indicators(data) + + +if __name__ == "__main__": + update_all_active()