JoinQuant 技术指标：MFI 资金流量指标策略实现指南

这是一个基于 MFI (Money Flow Index, 资金流量指标) 的量化交易策略。

策略原理

MFI 指标结合了价格和成交量，用于衡量买卖压力的强度。它的取值范围是 0 到 100。

• 超卖区域（买入信号）：当 MFI 低于 20 时，通常被认为市场处于超卖状态，价格可能反弹。
• 超买区域（卖出信号）：当 MFI 高于 80 时，通常被认为市场处于超买状态，价格可能回调。

策略逻辑

1. 股票池：选取沪深300成分股。
2. 买入条件：当某只股票的 14日 MFI 指标低于 20，且当前持仓未满，则买入。
3. 卖出条件：当持仓股票的 14日 MFI 指标高于 80，则卖出止盈；或者股票被调出沪深300成分股时卖出。
4. 仓位管理：最多持有 10 只股票，每只股票等权重配置。
5. 交易频率：按日回测，每天开盘前运行一次。

策略代码

# -*- coding: utf-8 -*-
import jqdata
import pandas as pd
import numpy as np

def initialize(context):
    """
    初始化函数，设定基准、手续费、滑点、全局变量等
    """
    # 设定沪深300作为基准
    set_benchmark('000300.XSHG')
    # 开启动态复权模式(真实价格)
    set_option('use_real_price', True)
    # 过滤掉order系列API产生的比error级别低的log
    log.set_level('order', 'error')
    
    # 设定股票交易手续费：买入万分之三，卖出万分之三加千分之一印花税，最低5元
    set_order_cost(OrderCost(close_tax=0.001, open_commission=0.0003, close_commission=0.0003, min_commission=5), type='stock')
    
    # 定义全局变量
    g.index_security = '000300.XSHG' # 股票池基准
    g.mfi_period = 14                # MFI计算周期
    g.buy_threshold = 20             # 买入阈值（超卖）
    g.sell_threshold = 80            # 卖出阈值（超买）
    g.max_holdings = 10              # 最大持仓数量
    
    # 每天开盘时运行
    run_daily(market_open, time='09:30')

def market_open(context):
    """
    每日交易逻辑
    """
    # 1. 获取股票池（沪深300）
    check_out_list = get_index_stocks(g.index_security)
    
    # 2. 卖出逻辑
    # 遍历当前持仓
    for security in list(context.portfolio.positions.keys()):
        # 获取持仓对象
        position = context.portfolio.positions[security]
        
        # 如果可用仓位为0，跳过
        if position.closeable_amount == 0:
            continue
            
        # 如果股票不在当期成分股中，卖出
        if security not in check_out_list:
            order_target_value(security, 0)
            log.info("股票 %s 调出成分股，卖出" % security)
            continue
            
        # 计算MFI
        mfi = calculate_mfi(security, g.mfi_period)
        
        # 如果 MFI > 80 (超买)，卖出止盈
        if mfi is not None and mfi > g.sell_threshold:
            order_target_value(security, 0)
            log.info("股票 %s MFI值为 %.2f，触发超买阈值，卖出" % (security, mfi))

    # 3. 买入逻辑
    # 获取当前持仓数量
    current_holdings = len(context.portfolio.positions)
    if current_holdings >= g.max_holdings:
        return # 持仓已满，不再买入

    # 计算剩余可买入的股票数量
    available_slots = g.max_holdings - current_holdings
    # 计算每只股票的目标资金（简单的等权重分配）
    # 注意：这里使用总资产来计算每份资金，也可以使用 available_cash / available_slots
    target_value_per_stock = context.portfolio.total_value / g.max_holdings
    
    buy_list = []
    
    # 遍历股票池寻找买入机会
    # 为了提高回测速度，这里可以先筛选一部分，或者直接遍历。
    # 实际策略中建议结合其他因子初筛，这里演示纯MFI逻辑。
    for security in check_out_list:
        # 如果已经持仓，跳过
        if security in context.portfolio.positions:
            continue
            
        mfi = calculate_mfi(security, g.mfi_period)
        
        # 如果 MFI < 20 (超卖)，加入待买入列表
        if mfi is not None and mfi < g.buy_threshold:
            buy_list.append((security, mfi))
    
    # 按照 MFI 从小到大排序（优先买入超卖最严重的）
    buy_list.sort(key=lambda x: x[1])
    
    # 执行买入
    for security, mfi_val in buy_list[:available_slots]:
        if context.portfolio.available_cash < target_value_per_stock:
            break # 现金不足
            
        order_target_value(security, target_value_per_stock)
        log.info("股票 %s MFI值为 %.2f，触发超卖阈值，买入" % (security, mfi_val))

def calculate_mfi(security, period=14):
    """
    计算 MFI 指标
    公式:
    1. 典型价格 (TP) = (High + Low + Close) / 3
    2. 资金流量 (MF) = TP * Volume
    3. 如果 当日TP > 昨日TP，则为正资金流量(PMF)，否则为负资金流量(NMF)
    4. 资金流量比率 (MFR) = sum(PMF, period) / sum(NMF, period)
    5. MFI = 100 - (100 / (1 + MFR))
    """
    # 获取历史数据：需要 period + 1 天的数据，因为要和前一天比较
    # 包含：最高价、最低价、收盘价、成交量
    data = attribute_history(security, period + 1, '1d', ['high', 'low', 'close', 'volume'], skip_paused=True)
    
    # 如果数据长度不足，无法计算
    if len(data) < period + 1:
        return None
        
    # 1. 计算典型价格 TP
    # pandas Series 操作
    tp = (data['high'] + data['low'] + data['close']) / 3.0
    
    # 2. 计算原始资金流量 Raw Money Flow
    raw_money_flow = tp * data['volume']
    
    # 3. 区分正负资金流量
    # 获取今日和昨日的TP进行比较 (从索引1开始到最后)
    # data长度为15，我们需要比较后14天相对于它们前一天的涨跌
    
    positive_flow = []
    negative_flow = []
    
    # 遍历后14天的数据
    for i in range(1, len(tp)):
        if tp[i] > tp[i-1]:
            positive_flow.append(raw_money_flow[i])
            negative_flow.append(0)
        elif tp[i] < tp[i-1]:
            positive_flow.append(0)
            negative_flow.append(raw_money_flow[i])
        else:
            # 价格不变，通常忽略或两边都为0
            positive_flow.append(0)
            negative_flow.append(0)
            
    # 4. 计算周期内的和
    sum_pos = sum(positive_flow)
    sum_neg = sum(negative_flow)
    
    # 5. 计算 MFI
    if sum_neg == 0:
        # 如果负向资金流为0，说明全是涨的，MFI为100
        return 100.0
    
    mfr = sum_pos / sum_neg
    mfi = 100 - (100 / (1 + mfr))
    
    return mfi

常见问题 Q&A

Q: 为什么选择 MFI 而不是 RSI？
A: RSI (相对强弱指标) 仅考虑价格变动，而 MFI 同时考虑了价格和成交量。在量化交易中，成交量往往能验证趋势的有效性。MFI 可以被视为“加权的 RSI”。

Q: 策略中的 attribute_history 获取的数据包含当天吗？
A: 在 run_daily 设置为 09:30 (开盘运行) 时，attribute_history 获取的是截止到昨天收盘的数据。这是为了避免使用“未来函数”（即在今天开盘时就知道了今天的收盘价）。

Q: 如何优化这个策略？
A:

1. 结合趋势指标：MFI 是反转指标，在强劲的单边趋势中可能会过早发出反转信号（钝化）。可以结合均线（MA）过滤，例如只有在股价位于 60日均线之上时才执行 MFI 超卖买入。
2. 动态阈值：根据大盘的波动率动态调整 20/80 的阈值。
3. 止损机制：当前代码仅依靠 MFI > 80 止盈，建议添加固定比例止损（如亏损 10% 止损）以控制风险。