沪深港通数据实战：从北向资金到择时信号的全链路量化（第 2 篇）：北向资金走势深度：累计净流入与自研动量信号

沪深港通数据实战：从北向资金到择时信号的全链路量化（第 2 篇）：北向资金走势深度：累计净流入与自研动量信号

一、前言

第 1 篇已经把 ht_bxls（日度北向/沪股通/深股通净流入）与 ht_bxzl（累计值）落库。本篇不再拉新接口，而是对已有时序做深加工：累计净流入曲线、日度净流入、N 日动量，以及一个可量化的择时阈值信号，为第 11 篇的综合择时指标提供「资金面」单因子。

代码保持自包含：仅从 quant.db 读取第 1 篇的数据，不依赖网络。

二、接口与字段（本篇复用第 1 篇落库表）

三、自研衍生指标

四、数据表设计

CREATE TABLE ht_bx_trend (
t TEXT, bx REAL, cum_bx REAL, daily_net REAL, mom_n REAL, signal INTEGER);

五、完整代码

import sqlite3, pandas as pd, logging
from datetime import datetime

DB_PATH = "quant.db"
LOG_FILE = "season4_article2.log"
logging.basicConfig(filename=LOG_FILE, level=logging.INFO,
format="%(asctime)s %(levelname)s %(message)s")

def build_trend(n=5):
"""基于 ht_bxls 计算累计/日度/动量并生成择时信号。"""
conn = sqlite3.connect(DB_PATH)
df = pd.read_sql("SELECT t, bx, hgt, sgt FROM ht_bxls ORDER BY t", conn)
conn.close()
df["t"] = pd.to_datetime(df["t"])
df = df.sort_values("t").reset_index(drop=True)
df["cum_bx"] = df["bx"].cumsum()
df["daily_net"] = df["bx"].diff()
df["mom_n"] = df["bx"].diff(n)
df["signal"] = 0
df.loc[(df["bx"] > 0) & (df["mom_n"] > 0), "signal"] = 1
df.loc[df["bx"] < 0, "signal"] = –1
conn = sqlite3.connect(DB_PATH)
conn.execute("""CREATE TABLE IF NOT EXISTS ht_bx_trend (
t TEXT, bx REAL, cum_bx REAL, daily_net REAL, mom_n REAL, signal INTEGER)""")
conn.execute("DELETE FROM ht_bx_trend")
for _, row in df.iterrows():
conn.execute("INSERT INTO ht_bx_trend VALUES (?,?,?,?,?,?)",
(row["t"].strftime("%Y-%m-%d"), row["bx"], row["cum_bx"],
row["daily_net"], row["mom_n"], int(row["signal"])))
conn.commit(); conn.close()
logging.info("trend built %d rows", len(df))
return df

def check_vs_bxzl():
"""自研累计与官方累计交叉校验。"""
conn = sqlite3.connect(DB_PATH)
last = pd.read_sql("SELECT cum_bx FROM ht_bx_trend ORDER BY t DESC LIMIT 1", conn)
allv = pd.read_sql("SELECT bxall FROM ht_bxzl ORDER BY ts DESC LIMIT 1", conn)
conn.close()
if not last.empty and not allv.empty:
print("自研累计=%.2f 官方累计=%.2f" % (last["cum_bx"].iloc[0], allv["bxall"].iloc[0]))

if __name__ == "__main__":
df = build_trend(n=5)
print(df.tail(10).to_string())
print("看多天数:", (df["signal"] == 1).sum(), " 看空天数:", (df["signal"] == –1).sum())
check_vs_bxzl()

六、业务关键点

• bx 单位依官方文档（通常为亿元），cumsum 与 bxall 完全可比的前提是 ht_bxls 已拉满全部历史交易日。
• 信号阈值可参数化：mom_n > 0 只是基线，实盘可换成「N 日动量 > 均线」或「连续 K 日为正」。
• 第 11 篇会把本篇 signal 与后续成分股涨跌、排名持续度做因子共振。

七、拓展练习

八、下篇预告

下一篇《沪、深股通成分股行情：字段解析与清洗》将接入 ht/nbzj/hgtc 与 ht/nbzj/sgtc，把北向通道的「标的池」行情落库，并计算北向成分股当日平均涨跌幅这一情绪指标。

免责申明：文中所有数据处理逻辑仅为编程演示，仅为数据演示，不构成投资建议。市场有风险，投资需谨慎。