feat: factor_engine集成mcap_q/pe_q分位加工——stock_fundamentals全市场排序bisect算横截面分位,存stock_indicators(不重复存PE/市值,只存衍生分位)

deploy/profile-scripts/factor_engine.py

@@ -49,8 +49,15 @@ def init_tables(conn):
             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,
+            mcap_q REAL, pe_q REAL,
             updated_at TEXT, PRIMARY KEY (code, date)
         )""")
+    # 兼容已存在的表（补 mcap_q/pe_q 列）
+    for col in ("mcap_q", "pe_q"):
+        try:
+            conn.execute(f"ALTER TABLE stock_indicators ADD COLUMN {col} REAL")
+        except Exception:
+            pass
     conn.execute("""
         CREATE TABLE IF NOT EXISTS market_indicators (
             date TEXT PRIMARY KEY,
@@ -61,6 +68,18 @@ def init_tables(conn):
     conn.commit()
 
 
+def load_fundamentals_sorted(conn):
+    """读 stock_fundamentals 全市场 mcap/pe 排序（算横截面分位用）。
+    返回 (mcap_sorted_list, pe_sorted_list, code→(mcap,pe) dict)。
+    mcap_q/pe_q 是衍生分位（由 mcap/pe 算），存加工层，不重复存原始 PE/市值。"""
+    import bisect
+    rows = conn.execute("SELECT code, mcap_total, pe FROM stock_fundamentals").fetchall()
+    mcaps = sorted(r[1] for r in rows if r[1] and r[1] > 0)
+    pes = sorted(r[2] for r in rows if r[2] and r[2] > 0)
+    code_map = {r[0]: (r[1], r[2]) for r in rows}
+    return mcaps, pes, code_map, bisect
+
+
 def calc_stock_indicators(code, bars):
     """由日K bars 算个股指标（bars: [{date,open,close,high,low,volume,amount}]，需>=65根）"""
     if len(bars) < 65:
@@ -191,6 +210,9 @@ def main():
     print(f"  股票池: {len(codes)} 只", flush=True)
 
     cur = conn.cursor()
+    # 横截面分位（mcap_q/pe_q，由 stock_fundamentals 全市场排序算）
+    mcaps, pes, fund_map, bisect = load_fundamentals_sorted(conn)
+    n_mcap, n_pe = len(mcaps), len(pes)
     ok = fail = skip = written = 0
     latest_date = None
     for idx, code in enumerate(codes, 1):
@@ -205,17 +227,27 @@ def main():
                 skip += 1
                 continue
             latest_date = ind["date"]
+            # 分位（bisect 查全市场排名）
+            mcap_q = pe_q = None
+            if code in fund_map:
+                mcap, pe = fund_map[code]
+                if mcap and mcap > 0 and n_mcap:
+                    mcap_q = round(bisect.bisect_left(mcaps, mcap) / n_mcap, 3)
+                if pe and pe > 0 and n_pe:
+                    pe_q = round(bisect.bisect_left(pes, pe) / n_pe, 3)
             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'))""",
+                 amount_ma20, atr_pct, close_up, trend_aligned, hh_structure, hl_structure,
+                 mcap_q, pe_q, 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"]))
+                 ind["close_up"], ind["trend_aligned"], ind["hh_structure"], ind["hl_structure"],
+                 mcap_q, pe_q))
             written += 1
             ok += 1
         except Exception as e: