JoinQuant聚宽量化平台：主力资金连续三天流入选股策略编写指南（附源码）

你好！你的思路非常正确，在 JoinQuant（聚宽）平台上，获取资金流向数据正是使用 get_money_flow 函数。

要实现“主力资金连续三天流入”的选股指标，我们需要关注 get_money_flow 返回数据中的 net_amount_main（主力净额）字段。当该字段大于 0 时，表示主力资金净流入。

下面为你提供一个完整的选股函数代码，以及如何在一个简单的策略框架中调用它。

1. 核心选股逻辑解析

• API 使用：get_money_flow(security_list, end_date, fields=['sec_code', 'net_amount_main'], count=3)
• 判断条件：按股票分组后，判断过去 3 个交易日的 net_amount_main 是否全部大于 0。

2. 完整策略源码示例

你可以将以下代码直接复制到聚宽的策略研究或回测环境中运行：

# 导入聚宽函数库
import jqdata
import pandas as pd

# 初始化函数，设定基准等等
def initialize(context):
    # 设定沪深300作为基准
    set_benchmark('000300.XSHG')
    # 开启动态复权模式(真实价格)
    set_option('use_real_price', True)
    # 过滤掉order系列API产生的比error级别低的log
    log.set_level('order', 'error')
    
    # 每天开盘前运行选股逻辑
    run_daily(before_market_open, time='09:00', reference_security='000300.XSHG')

# 自定义选股函数：主力资金连续三天流入
def get_main_money_in_stocks(context, stock_list):
    """
    筛选主力资金连续三天流入的股票
    :param context: 策略上下文
    :param stock_list: 待筛选的股票池列表
    :return: 符合条件的股票列表
    """
    # 获取过去3天的资金流向数据
    # 注意：为了避免未来函数，end_date 使用 context.previous_date (昨天)
    df = get_money_flow(stock_list, 
                        end_date=context.previous_date, 
                        fields=['date', 'sec_code', 'net_amount_main'], 
                        count=3)
    
    selected_stocks = []
    
    # 如果获取不到数据，直接返回空列表
    if df is None or df.empty:
        return selected_stocks
        
    # 按股票代码进行分组
    grouped = df.groupby('sec_code')
    
    for code, group in grouped:
        # 确保取到了完整的3天数据
        if len(group) == 3:
            # 判断这3天的主力净额是否都大于0
            if all(group['net_amount_main'] > 0):
                selected_stocks.append(code)
                
    return selected_stocks

# 开盘前运行函数
def before_market_open(context):
    # 1. 设定初始股票池，这里以沪深300成分股为例
    # 你也可以换成 get_all_securities(['stock']).index 获取全市场股票
    initial_pool = get_index_stocks('000300.XSHG')
    
    # 2. 过滤停牌的股票 (可选，但推荐)
    current_data = get_current_data()
    tradeable_pool = [stock for stock in initial_pool if not current_data[stock].paused]
    
    # 3. 调用选股函数，获取主力连续3天流入的股票
    g.target_list = get_main_money_in_stocks(context, tradeable_pool)
    
    log.info(f"今日选出主力连续3天流入的股票共 {len(g.target_list)} 只:")
    log.info(g.target_list)

# 开盘时运行函数 (这里可以编写你的买卖逻辑)
def handle_data(context, data):
    # 示例：简单地将选出的股票等权买入
    if not g.target_list:
        return
        
    # 卖出不在目标列表中的持仓
    for stock in list(context.portfolio.positions.keys()):
        if stock not in g.target_list:
            order_target(stock, 0)
            
    # 买入目标列表中的股票
    cash_per_stock = context.portfolio.available_cash / len(g.target_list)
    for stock in g.target_list:
        if stock not in context.portfolio.positions:
            order_value(stock, cash_per_stock)

3. 代码细节提示

1. 避免未来函数：在调用 get_money_flow 时，end_date 参数务必使用 context.previous_date（即前一个交易日）。因为当天的资金流向数据在盘中是无法获取完整的，必须等收盘后结算。
2. 数据完整性校验：代码中使用了 if len(group) == 3: 来确保这只股票确实有 3 天的交易数据（比如刚上市的新股或长期停牌的股票可能数据不足）。
3. 性能优化：get_money_flow 支持传入股票列表批量获取数据，这比用 for 循环单只股票获取要快得多。代码中使用了 pandas 的 groupby 方法进行高效的数据处理。

你可以将这个 get_main_money_in_stocks 函数作为一个独立的模块，结合你自己的其他指标（如均线、市盈率等）组合使用，构建更强大的量化策略！