#!/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()