如何利用资金流数据构建选股策略？JoinQuant Python 实例

这是一个基于**主力资金净流入占比（net_pct_main）**进行选股的策略。

策略逻辑说明

1. 核心因子：使用聚宽 get_money_flow API 获取资金流数据。
   • 主力净流入：定义为超大单（特大单）和大单的净流入之和。
   • 主力净占比 (net_pct_main)：主力净流入金额 / 总成交额。使用占比而非绝对金额，可以避免大市值股票天然资金量大的偏差，更能反映资金买入的强度。
2. 股票池：选取沪深300成分股（流动性好，数据相对稳定）。
3. 过滤条件：
   • 剔除停牌股票。
   • 剔除ST/ST*股票。
   • 剔除涨跌停股票（无法买入或卖出）。
4. 选股逻辑：
   • 获取前一交易日的主力净占比数据。
   • 按照主力净占比从大到小排序。
   • 选取前 10 只股票。
5. 交易执行：
   • 每日开盘时调仓。
   • 卖出不在目标列表中的持仓。
   • 买入目标列表中的股票，等权重分配资金。

策略代码

# -*- coding: utf-8 -*-
# 导入函数库
from jqdata import *

def initialize(context):
    """
    初始化函数，设定基准、手续费、滑点、全局变量等
    """
    # 设定沪深300作为基准
    set_benchmark('000300.XSHG')
    # 开启动态复权模式(真实价格)
    set_option('use_real_price', True)
    # 过滤掉order系列API产生的比error级别低的log
    log.set_level('order', 'error')
    
    # --- 策略参数设置 ---
    # 持仓股票数量
    g.stock_num = 10
    # 股票池：沪深300
    g.index_scope = '000300.XSHG'
    
    # --- 费率设置 ---
    # 股票类每笔交易时的手续费是：买入时佣金万分之三，卖出时佣金万分之三加千分之一印花税, 每笔交易佣金最低扣5块钱
    set_order_cost(OrderCost(close_tax=0.001, open_commission=0.0003, close_commission=0.0003, min_commission=5), type='stock')
    
    # --- 运行频率 ---
    # 每天开盘时运行选股和交易
    run_daily(rebalance, '09:30')

def rebalance(context):
    """
    每日调仓主函数
    """
    # 1. 获取股票池（沪深300）
    check_out_list = get_index_stocks(g.index_scope)
    
    # 2. 过滤停牌、ST、涨跌停股票
    check_out_list = filter_basic_stock(context, check_out_list)
    
    # 3. 获取资金流数据并排序
    # 获取前一交易日的资金流数据，避免未来函数
    # net_pct_main: 主力净占比 (%) = (超大单净额 + 大单净额) / 成交额
    q = query(
        money_flow.code,
        money_flow.net_pct_main
    ).filter(
        money_flow.code.in_(check_out_list),
        money_flow.date == context.previous_date  # 取前一天的数据
    ).order_by(
        money_flow.net_pct_main.desc() # 按主力净占比降序排列
    ).limit(
        g.stock_num # 取前N只
    )
    
    df = get_money_flow(check_out_list, end_date=context.previous_date, fields=['net_pct_main'], count=1)
    
    # 注意：get_money_flow 返回的是 panel 或者 dataframe，处理方式如下
    # 这里我们直接处理 DataFrame，索引是日期，列是股票代码，或者直接用 query 方式更灵活
    # 为了演示 get_money_flow 的标准用法，我们重新通过 API 获取数据
    # 上面的 query 是查数据库方式，下面演示 API 方式，两者均可，API 方式更简单
    
    # 获取前一天的资金流数据
    # net_pct_main: 主力净占比
    mf_data = get_money_flow(check_out_list, end_date=context.previous_date, fields=['net_pct_main'], count=1)
    
    # 如果没有数据，直接返回
    if mf_data is None or len(mf_data) == 0:
        return

    # 提取数据，Series格式：index为股票代码，value为主力净占比
    # mf_data 索引是日期，我们需要取最后一行（即昨天）
    series_data = mf_data.iloc[0]
    
    # 剔除数据为 NaN 的股票
    series_data = series_data.dropna()
    
    # 排序：从大到小
    sorted_series = series_data.sort_values(ascending=False)
    
    # 选取前 g.stock_num 只股票
    target_list = list(sorted_series.index[:g.stock_num])
    
    log.info("今日目标持仓: %s" % target_list)
    
    # 4. 执行交易
    do_trade(context, target_list)

def filter_basic_stock(context, stock_list):
    """
    过滤停牌、ST、涨跌停股票
    """
    current_data = get_current_data()
    return [stock for stock in stock_list 
            if not current_data[stock].paused  # 未停牌
            and not current_data[stock].is_st  # 非ST
            and 'ST' not in current_data[stock].name 
            and '*' not in current_data[stock].name
            and current_data[stock].low_limit < current_data[stock].day_open < current_data[stock].high_limit # 开盘未涨跌停
           ]

def do_trade(context, target_list):
    """
    执行买卖操作
    """
    # 获取当前持仓
    current_holdings = list(context.portfolio.positions.keys())
    
    # 1. 卖出不在目标列表中的股票
    for stock in current_holdings:
        if stock not in target_list:
            order_target_value(stock, 0)
            log.info("卖出: %s" % stock)
            
    # 2. 买入目标列表中的股票
    # 计算每只股票应该分配的资金（等权重）
    # 注意：这里简单处理，每次全仓分配。实际中可能需要预留现金。
    if len(target_list) > 0:
        # 获取当前可用资金 + 刚刚卖出股票获得的资金（近似计算总资产）
        # 使用 total_value 更稳定，防止因卖出未到账导致买入资金不足
        position_count = len(target_list)
        value_per_stock = context.portfolio.total_value / position_count
        
        for stock in target_list:
            order_target_value(stock, value_per_stock)
            log.info("买入/调仓: %s, 目标金额: %f" % (stock, value_per_stock))

关键点解析

1. get_money_flow API:

   • 这是聚宽专门用于获取资金流向数据的接口。
   • fields=['net_pct_main']：我们选择了主力净占比。如果你想用绝对金额，可以改为 net_amount_main。
   • end_date=context.previous_date：在回测中，为了避免未来函数（即今天还没收盘就用今天的全天资金流数据），我们必须使用前一个交易日的数据来进行决策。

2. 数据排序:

   • 代码中使用了 sort_values(ascending=False) 对主力净占比进行降序排列，意味着我们买入的是主力资金流入意愿最强的股票。

3. 风险控制:

   • filter_basic_stock 函数非常重要，它剔除了停牌、ST 以及开盘即涨跌停的股票，保证了策略在实盘中的可执行性。

常见问题 (Q&A)

Q: 为什么使用主力净占比(net_pct_main)而不是主力净额(net_amount_main)？
A: 主力净额容易受市值影响，大盘股（如银行、两桶油）成交量巨大，即使主力流入比例很小，净额也可能很大。使用占比可以消除市值的规模影响，更准确地衡量资金对该股票的“青睐程度”。

Q: 这个策略适合什么行情？
A: 资金流策略通常属于趋势跟随策略。在市场情绪高涨、资金活跃的牛市或震荡市中表现较好。在单边下跌的熊市中，由于主力资金可能也在撤退，或者流入的股票具有补跌风险，策略可能面临回撤。

Q: 如何改进这个策略？
A:

1. 增加技术指标过滤：例如只买入股价在20日均线之上的股票，避免接飞刀。
2. 结合基本面：在资金流入的基础上，叠加低PE或高增长的因子。
3. 止损机制：增加个股止损（如亏损10%卖出）或大盘止损（如大盘跌破60日线空仓）。