diff --git a/deploy/profile-scripts/factor_engine.py b/deploy/profile-scripts/factor_engine.py new file mode 100644 index 00000000..37fe60f1 --- /dev/null +++ b/deploy/profile-scripts/factor_engine.py @@ -0,0 +1,252 @@ +#!/usr/bin/env python3 +"""factor_engine.py — 技术指标加工(数据加工层核心) + +背景(2026-08-12 架构补缺·加工层): + 技术指标(bias60/RSI/ADX/mom20/ma/macd/atr 等)API 拉不了,必须自己算—— + 此前分散在 5-6 个脚本现算(口径漂移风险)。本脚本统一加工: + 读 stock_daily → 算指标 → 写 stock_indicators/market_indicators, + 使用层(扫描器/重评/回测)直接读,不现算,口径完全一致。 + +算法权威源:backtest_framework.calc_*(回测体系,已验证)。实盘/回测共用同一套, + 消除 backtest_framework vs indicators 两套定义并存的口径分裂。 + +原则(老莫定):能拉取的拉取(PE/市值已由采集层拉),拉不了的才自己算(技术指标)。 + mcap_q/pe_q 分位由 stock_fundamentals 算(加工层),不重复存 PE/PB/市值。 + +调度:收盘后 5 17 * * 1-5(stock_daily 采集 16:35 完成后) +规范:单例守卫(5.3) + INSERT OR REPLACE 幂等 + 增量(只写最近1天)+ pandas 向量化 +""" +import sys, os, sqlite3, fcntl, time +from pathlib import Path +from datetime import datetime, timedelta + +sys.path.insert(0, "/home/hmo/MoFin") +from backtest_framework import (calc_ma, calc_rsi, calc_macd, calc_atr, + calc_trend_strength, calc_obv, calc_roc) + +DB_PATH = Path("/home/hmo/MoFin/data/mofin.db") +LOOKBACK_DAYS = 130 # 算 ma60/rsi/adx 需要的历史窗口(取130天保险) + + +def _singleton_guard(tag="factor_engine.py"): + lock_dir = Path("/tmp/mofin_locks") + lock_dir.mkdir(exist_ok=True) + try: + fd = os.open(str(lock_dir / f"{tag}.lock"), os.O_CREAT | os.O_RDWR) + fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB) + return fd + except OSError: + print(f"[{tag}] 已有实例在运行,退出", flush=True) + sys.exit(0) + + +def init_tables(conn): + conn.execute(""" + CREATE TABLE IF NOT EXISTS stock_indicators ( + code TEXT, date TEXT, + ma5 REAL, ma10 REAL, ma20 REAL, ma60 REAL, + rsi REAL, adx REAL, macd_hist REAL, atr REAL, roc REAL, obv REAL, + bias60 REAL, mom20 REAL, prev_ret60 REAL, dist_ma20 REAL, dist_lo20 REAL, r5f REAL, + vol_ratio REAL, vol_shrink REAL, amount_ma20 REAL, atr_pct REAL, + close_up INTEGER, trend_aligned INTEGER, hh_structure INTEGER, hl_structure INTEGER, + updated_at TEXT, PRIMARY KEY (code, date) + )""") + conn.execute(""" + CREATE TABLE IF NOT EXISTS market_indicators ( + date TEXT PRIMARY KEY, + mkt_rsi REAL, mkt_dd60 REAL, mkt_adx REAL, mkt_above_ma20 INTEGER, + mkt_down_days INTEGER, mkt_slope REAL, mkt_roc REAL, + updated_at TEXT + )""") + conn.commit() + + +def calc_stock_indicators(code, bars): + """由日K bars 算个股指标(bars: [{date,open,close,high,low,volume,amount}],需>=65根)""" + if len(bars) < 65: + return None + closes = [b["close"] for b in bars] + highs = [b["high"] for b in bars] + lows = [b["low"] for b in bars] + vols = [b["volume"] for b in bars] + amts = [b.get("amount") or 0 for b in bars] + i = len(bars) - 1 # 最后一根(当天) + close = closes[i] + if close <= 0: + return None + + ma5 = calc_ma(closes, 5); ma10 = calc_ma(closes, 10); ma20 = calc_ma(closes, 20); ma60 = calc_ma(closes, 60) + rsi = calc_rsi(closes) + _, _, macd_hist = calc_macd(closes) + atr = calc_atr(highs, lows, closes) + adx = calc_trend_strength(highs, lows, closes) + obv = calc_obv(closes, vols) + roc = calc_roc(closes) + + def g(arr, idx): + return arr[idx] if arr and idx < len(arr) and arr[idx] is not None else None + + ma20_v, ma60_v = g(ma20, i), g(ma60, i) + atr_v = g(atr, i) + bias60 = round((close - ma60_v) / ma60_v * 100, 2) if ma60_v and ma60_v > 0 else None + dist_ma20 = round((close - ma20_v) / ma20_v * 100, 2) if ma20_v and ma20_v > 0 else None + mom20 = round((close - closes[i-20]) / closes[i-20] * 100, 2) if i >= 20 and closes[i-20] > 0 else None + prev_ret60 = round((close - closes[i-60]) / closes[i-60] * 100, 2) if i >= 60 and closes[i-60] > 0 else None + r5f = round((close - closes[i-5]) / closes[i-5] * 100, 2) if i >= 5 and closes[i-5] > 0 else None + lo20 = min(lows[i-19:i+1]) if i >= 19 else None + dist_lo20 = round((close - lo20) / lo20 * 100, 2) if lo20 and lo20 > 0 else None + pvol = vols[i-5] if i >= 5 and vols[i-5] > 0 else None + vol_ratio = round(vols[i] / pvol, 2) if pvol else None + vol5 = sum(vols[i-4:i+1]) / 5 if i >= 4 else None + vol20 = sum(vols[i-19:i+1]) / 20 if i >= 19 else None + vol_shrink = round(vol5 / vol20, 2) if vol5 and vol20 and vol20 > 0 else None + amt_valid = [a for a in amts[i-19:i+1] if a > 0] if i >= 19 else [] + amount_ma20 = round(sum(amt_valid) / len(amt_valid) / 1000.0, 2) if amt_valid else None # 千元→百万 + atr_pct = round(atr_v / close * 100, 2) if atr_v and close > 0 else None + close_up = 1 if i >= 1 and close > closes[i-1] else 0 + trend_aligned = 1 if (ma5[i] and ma10[i] and ma20[i] and ma5[i] > ma10[i] > ma20[i] > 0) else 0 + hh = hl = 0 + if i >= 10: + h5 = max(highs[i-4:i+1]); h10 = max(highs[i-9:i-4]) + l5 = min(lows[i-4:i+1]); l10 = min(lows[i-9:i-4]) + hh = 1 if h5 > h10 else 0 + hl = 1 if l5 > l10 else 0 + + return { + "code": code, "date": bars[i]["date"], + "ma5": g(ma5, i), "ma10": g(ma10, i), "ma20": ma20_v, "ma60": ma60_v, + "rsi": g(rsi, i), "adx": g(adx, i), "macd_hist": g(macd_hist, i), + "atr": atr_v, "roc": g(roc, i), "obv": g(obv, i), + "bias60": bias60, "mom20": mom20, "prev_ret60": prev_ret60, + "dist_ma20": dist_ma20, "dist_lo20": dist_lo20, "r5f": r5f, + "vol_ratio": vol_ratio, "vol_shrink": vol_shrink, "amount_ma20": amount_ma20, + "atr_pct": atr_pct, "close_up": close_up, "trend_aligned": trend_aligned, + "hh_structure": hh, "hl_structure": hl, + } + + +def calc_market_indicators(bars): + """由大盘指数日K算大盘指标(mkt_rsi/mkt_dd60/mkt_adx/mkt_above_ma20/mkt_down_days/mkt_slope/mkt_roc)""" + if len(bars) < 65: + return None + closes = [b["close"] for b in bars] + highs = [b["high"] for b in bars] + lows = [b["low"] for b in bars] + i = len(bars) - 1 + close = closes[i] + ma20 = calc_ma(closes, 20) + rsi = calc_rsi(closes) + adx = calc_trend_strength(highs, lows, closes) + roc = calc_roc(closes) + + def g(arr, idx): + return arr[idx] if arr and idx < len(arr) and arr[idx] is not None else None + + ma20_v = g(ma20, i) + mkt_above = 1 if (ma20_v and close > ma20_v) else 0 + hi60 = max(highs[max(0, i-59):i+1]) + mkt_dd60 = round((close - hi60) / hi60 * 100, 2) if hi60 > 0 else None + # 连跌天数 + down = 0 + for k in range(i, 0, -1): + if closes[k] < closes[k-1]: + down += 1 + else: + break + slope = None + if i >= 5 and g(ma20, i) and g(ma20, i-5) and g(ma20, i-5) > 0: + slope = round((g(ma20, i) - g(ma20, i-5)) / g(ma20, i-5) * 100, 3) + return { + "date": bars[i]["date"], "mkt_rsi": g(rsi, i), "mkt_dd60": mkt_dd60, + "mkt_adx": g(adx, i), "mkt_above_ma20": mkt_above, "mkt_down_days": down, + "mkt_slope": slope, "mkt_roc": g(roc, i), + } + + +def load_bars(conn, code, since): + rows = conn.execute( + "SELECT date, open, close, high, low, volume, amount FROM stock_daily " + "WHERE code=? AND date>=? ORDER BY date", (code, since)).fetchall() + return [{"date": r[0], "open": r[1], "close": r[2], "high": r[3], "low": r[4], + "volume": r[5], "amount": r[6]} for r in rows] + + +def main(): + _fd = _singleton_guard() + t0 = time.time() + print(f"[factor_engine] {datetime.now().strftime('%H:%M:%S')} 技术指标加工开始", flush=True) + + conn = sqlite3.connect(str(DB_PATH), timeout=60) + init_tables(conn) + since = (datetime.now() - timedelta(days=LOOKBACK_DAYS * 2)).strftime("%Y-%m-%d") + codes = [str(r[0]) for r in conn.execute( + "SELECT DISTINCT code FROM stock_daily WHERE length(code)=6 ORDER BY code").fetchall()] + limit = 0 + for a in sys.argv[1:]: + if a.startswith("--limit"): + limit = int(a.split("=")[-1] if "=" in a else sys.argv[sys.argv.index(a) + 1]) + if limit > 0: + codes = codes[:limit] + print(f" [测试模式] 只跑前 {limit} 只", flush=True) + print(f" 股票池: {len(codes)} 只", flush=True) + + cur = conn.cursor() + ok = fail = skip = written = 0 + latest_date = None + for idx, code in enumerate(codes, 1): + try: + bars = load_bars(conn, code, since) + if len(bars) < LOOKBACK_DAYS: + skip += 1 + continue + bars = bars[-LOOKBACK_DAYS:] + ind = calc_stock_indicators(code, bars) + if not ind: + skip += 1 + continue + latest_date = ind["date"] + cur.execute(""" + INSERT OR REPLACE INTO stock_indicators + (code, date, ma5, ma10, ma20, ma60, rsi, adx, macd_hist, atr, roc, obv, + bias60, mom20, prev_ret60, dist_ma20, dist_lo20, r5f, vol_ratio, vol_shrink, + amount_ma20, atr_pct, close_up, trend_aligned, hh_structure, hl_structure, updated_at) + VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,datetime('now','localtime'))""", + (ind["code"], ind["date"], ind["ma5"], ind["ma10"], ind["ma20"], ind["ma60"], + ind["rsi"], ind["adx"], ind["macd_hist"], ind["atr"], ind["roc"], ind["obv"], + ind["bias60"], ind["mom20"], ind["prev_ret60"], ind["dist_ma20"], ind["dist_lo20"], + ind["r5f"], ind["vol_ratio"], ind["vol_shrink"], ind["amount_ma20"], ind["atr_pct"], + ind["close_up"], ind["trend_aligned"], ind["hh_structure"], ind["hl_structure"])) + written += 1 + ok += 1 + except Exception as e: + fail += 1 + if fail <= 5: + print(f" FAIL {code}: {str(e)[:60]}", flush=True) + if idx % 200 == 0: + conn.commit() + print(f" [{idx}/{len(codes)}] ok={ok} skip={skip} fail={fail} | {time.time()-t0:.0f}s", flush=True) + conn.commit() + + # 大盘指标(上证指数 sh000001) + try: + mbars = load_bars(conn, "sh000001", since)[-LOOKBACK_DAYS:] + mind = calc_market_indicators(mbars) + if mind: + cur.execute(""" + INSERT OR REPLACE INTO market_indicators + (date, mkt_rsi, mkt_dd60, mkt_adx, mkt_above_ma20, mkt_down_days, mkt_slope, mkt_roc, updated_at) + VALUES (?,?,?,?,?,?,?,?,datetime('now','localtime'))""", + (mind["date"], mind["mkt_rsi"], mind["mkt_dd60"], mind["mkt_adx"], + mind["mkt_above_ma20"], mind["mkt_down_days"], mind["mkt_slope"], mind["mkt_roc"])) + conn.commit() + print(f" 大盘指标: {mind['date']} rsi={mind['mkt_rsi']} dd60={mind['mkt_dd60']} adx={mind['mkt_adx']}", flush=True) + except Exception as e: + print(f" 大盘指标失败: {e}", flush=True) + + conn.close() + dt = time.time() - t0 + print(f"[factor_engine] 完成: {written} 只写入(最新 {latest_date}), skip={skip} fail={fail}, 耗时 {dt:.0f}s", flush=True) + + +if __name__ == "__main__": + main()