量化数据库个股资金流向数据获取全框架详解stock_moneyflow

化数据库个股资金流向数据获取全框架详解 以 stock_moneyflow 接口为例，我为你拆解五大主流框架的数据获取方式、应用场景和完整代码实现。

详细数据在量化因子库中，需要的朋友找我领取：

一、数据获取核心函数分类

框架类型	数据获取方式	典型函数/方法	特点
回测框架	内置数据源	bt.feeds.PandasData()	历史数据一次性加载
实盘框架	实时订阅	xtdata.subscribe()	流式数据推送
投研框架	数据查询	jqdatasdk.get_money_flow()	灵活查询，分析导向
事件驱动	事件响应	handle_data()	按时间触发
混合框架	多源整合	DataLoader.load()	统一接口

---

二、各框架详细实现方案

框架1：Backtrader（最流行回测框架）

自定义资金流向数据源

class MoneyFlowData(bt.feeds.PandasData):

lines = ('net_buy', 'main_buy', 'retail_sell')  # 自定义字段
params = (
    ('net_buy', -1),  # 字段在DataFrame中的列索引
    ('main_buy', -1),
    ('retail_sell', -1),
)

获取数据函数

def get_stock_moneyflow(stock_code, start_date, end_date):

"""
从数据库或API获取资金流向数据
"""
# 模拟从数据库查询
import pymysql
conn = pymysql.connect(host='localhost', db='quant_db')
sql = f"""
    SELECT date, close, net_buy, main_buy, retail_sell 
    FROM stock_moneyflow 
    WHERE code='{stock_code}' 
    AND date BETWEEN '{start_date}' AND '{end_date}'
"""
df = pd.read_sql(sql, conn, index_col='date', parse_dates=True)
conn.close()

return df

策略中应用

class MoneyFlowStrategy(bt.Strategy):

params = dict(
    mf_period=20,  # 资金流向计算周期
    threshold=1.5  # 资金流入强度阈值
)

def __init__(self):
    self.dataclose = self.datas[0].close
    self.mf = self.datas[0].net_buy  # 使用自定义资金流向字段

def next(self):
    # 计算N日主力资金净流入
    mf_ma = bt.indicators.SMA(self.mf, period=self.p.mf_period)

    # 策略逻辑：主力持续流入且股价突破
    if self.mf[0] > mf_ma[0] * self.p.threshold:
        if self.dataclose[0] > self.dataclose[-1] * 1.02:
            self.buy(size=1000)

    # 资金大幅流出时止损
    if self.mf[0] < mf_ma[0] * 0.5:
        if self.position:
            self.sell(size=self.position.size)

主程序

# 主程序 cerebro = bt.Cerebro() df = get_stock_moneyflow('600519.SH', '20230101', '20231231') data = MoneyFlowData(dataname=df) cerebro.adddata(data) cerebro.addstrategy(MoneyFlowStrategy, mf_period=20, threshold=1.5) cerebro.run()

应用场景

策略类型：主力追踪策略、资金流突破策略

时间周期：日线及以上

标的：全市场个股筛选 优化技巧 Python 复制

批量加载多只股票

def load_multiple_stocks(codes, start, end):

for code in codes:
    df = get_stock_moneyflow(code, start, end)
    if not df.empty:
        cerebro.adddata(MoneyFlowData(dataname=df), name=code)

风险提示：

此内容不做任何参考，数据说明及风险告知

数据说明：

1、数据来源于tushare、qmt、同花顺、东方财富、开盘啦等多个渠道，版权归属于各平台；

2、数据准确性由平台负责，我方不对数据做任何保证；

3、使用请遵守平台方的使用规范要求；

4、请自行向相关平台方付费购买使用权限；联系我获取优惠购买渠道。

5、内容参考学习，请勿用于商业用途，也不构成投资建议，使用风险自负。

彩蛋福利：

1、QMT开通流程

（1）开户➕入金10万以上

（2）提供测试账号、安装配置流程、量化会员资料。

2、VIP服务

3、VIP福利

惊喜佣金

添加量化讨论群

量化工具全套资料库

提供服务器托管模式：ptrade策略+ldp极速柜台+vip定向服务器(限50人)