PTrade 量价分析：MFI 资金流量指标策略开发

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

策略说明

1. 核心逻辑：
   • MFI 是一种结合了价格和成交量的动量指标。
   • 买入信号：当 MFI 值低于 20 时，视为超卖（Oversold），市场可能反弹，执行买入。
   • 卖出信号：当 MFI 值高于 80 时，视为超买（Overbought），市场可能回调，执行卖出。
2. MFI 计算方法：
   • 由于 PTrade API 文档中未直接提供 get_MFI 函数，本策略包含了一个自定义的 calculate_mfi 函数，利用 get_history 获取的高、低、收、成交量数据进行计算。
3. 运行频率：日线级别（也可以修改为分钟级别）。
4. 标的：示例中使用 贵州茅台 (600519.SS)，你可以根据需要在 initialize 中修改。

策略源码

import numpy as np
import pandas as pd

def initialize(context):
    """
    初始化函数，设置策略参数和股票池
    """
    # 设置要操作的股票，这里以贵州茅台为例
    g.security = '600519.SS'
    set_universe(g.security)
    
    # MFI 指标参数
    g.mfi_period = 14      # 计算周期，通常为14
    g.buy_threshold = 20   # 超卖阈值，低于此值买入
    g.sell_threshold = 80  # 超买阈值，高于此值卖出
    
    # 设置回测费用（可选，为了更接近真实回测）
    set_commission(commission_ratio=0.0003, min_commission=5.0, type="STOCK")
    set_slippage(slippage=0.002)

def calculate_mfi(high, low, close, volume, n=14):
    """
    自定义 MFI 计算函数
    :param high: 最高价序列 (Series)
    :param low: 最低价序列 (Series)
    :param close: 收盘价序列 (Series)
    :param volume: 成交量序列 (Series)
    :param n: 周期
    :return: MFI 值序列
    """
    # 1. 计算典型价格 (Typical Price)
    tp = (high + low + close) / 3.0
    
    # 2. 计算原始资金流量 (Raw Money Flow)
    rmf = tp * volume
    
    # 3. 区分正向资金流和负向资金流
    # 获取典型价格的差值
    delta = tp.diff()
    
    # 正向流：当前TP > 昨日TP，取RMF，否则为0
    pos_flow = pd.Series(0.0, index=tp.index)
    pos_flow[delta > 0] = rmf[delta > 0]
    
    # 负向流：当前TP < 昨日TP，取RMF，否则为0
    neg_flow = pd.Series(0.0, index=tp.index)
    neg_flow[delta < 0] = rmf[delta < 0]
    
    # 4. 计算周期内的资金流汇总 (Rolling Sum)
    # PTrade pandas版本通常支持 rolling
    pos_mf_sum = pos_flow.rolling(window=n).sum()
    neg_mf_sum = neg_flow.rolling(window=n).sum()
    
    # 5. 计算资金流量比率 (Money Flow Ratio)
    # 处理除以0的情况
    mfr = pos_mf_sum / neg_mf_sum
    
    # 6. 计算 MFI
    mfi = 100 - (100 / (1 + mfr))
    
    return mfi

def handle_data(context, data):
    """
    盘中逻辑处理函数，每个单位时间（如每天）运行一次
    """
    security = g.security
    
    # 获取历史数据
    # 我们需要计算14日的MFI，为了保证rolling计算有足够的数据，
    # 我们获取 n + 20 天的数据，防止数据不足导致NaN
    fetch_len = g.mfi_period + 20
    
    # 获取日线数据：最高、最低、收盘、成交量
    # 注意：get_history 返回的是 DataFrame，索引是时间
    hist = get_history(fetch_len, frequency='1d', field=['high', 'low', 'close', 'volume'], security_list=security, fq='pre')
    
    # 检查数据长度是否足够
    if len(hist) < g.mfi_period + 1:
        log.info("数据不足，跳过计算")
        return

    # 提取各列数据
    high = hist['high']
    low = hist['low']
    close = hist['close']
    volume = hist['volume']
    
    # 计算 MFI
    mfi_series = calculate_mfi(high, low, close, volume, n=g.mfi_period)
    
    # 获取当天的 MFI 值 (序列的最后一个值)
    current_mfi = mfi_series.iloc[-1]
    
    # 如果计算结果是 NaN (例如停牌或刚上市)，则跳过
    if np.isnan(current_mfi):
        return
        
    # 获取当前持仓
    position = get_position(security).amount
    cash = context.portfolio.cash
    current_price = data[security]['close']
    
    # 打印日志方便调试
    # log.info("日期: %s, MFI: %.2f, 持仓: %s" % (context.blotter.current_dt, current_mfi, position))
    
    # --- 交易逻辑 ---
    
    # 1. 买入逻辑：MFI < 20 (超卖) 且 当前无持仓
    if current_mfi < g.buy_threshold and position == 0:
        log.info("MFI超卖 (%.2f < %s)，触发买入: %s" % (current_mfi, g.buy_threshold, security))
        # 全仓买入
        order_value(security, cash)
        
    # 2. 卖出逻辑：MFI > 80 (超买) 且 有持仓
    elif current_mfi > g.sell_threshold and position > 0:
        log.info("MFI超买 (%.2f > %s)，触发卖出: %s" % (current_mfi, g.sell_threshold, security))
        # 清仓卖出
        order_target(security, 0)

代码详解

1. initialize(context):

   • 设定了股票池为 600519.SS。
   • 定义了 MFI 的关键参数：周期 14，买入阈值 20，卖出阈值 80。
   • 设置了手续费和滑点，这对于回测结果的真实性很重要。

2. calculate_mfi(...):

   • 这是策略的核心。因为 PTrade 没有内置 MFI 函数，我们使用 pandas 库手动实现。
   • 典型价格 (TP)：(最高价 + 最低价 + 收盘价) / 3。
   • 资金流 (Money Flow)：典型价格 * 成交量。
   • 通过比较当日 TP 和昨日 TP，将资金流分为正向和负向。
   • 使用 rolling(window=n).sum() 计算过去 N 天的资金流总和。
   • 最后根据公式 MFI = 100 - 100 / (1 + MFR) 算出指标。

3. handle_data(context, data):

   • 使用 get_history 获取所需的 K 线数据（High, Low, Close, Volume）。
   • 调用计算函数得到当前的 MFI 值。
   • 买入：当 MFI 小于 20 且空仓时，使用 order_value 全仓买入。
   • 卖出：当 MFI 大于 80 且持有股票时，使用 order_target 清仓卖出。

注意事项

• 数据长度：在调用 get_history 时，获取的数据长度（fetch_len）必须大于 MFI 的计算周期，否则计算结果会全是 NaN。代码中设置为 g.mfi_period + 20 以确保安全。
• Python 版本：代码使用了兼容 Python 3.5+ 的语法（例如使用 %s 格式化字符串而不是 f-string），以确保在 PTrade 的旧版环境中也能运行。
• 多股操作：如果需要操作多只股票，可以在 initialize 中将 g.security 设置为列表，并在 handle_data 中使用 for 循环遍历处理。

Q&A

Q: 如何修改策略为操作多只股票？
A: 在 initialize 中设置 g.security = ['600519.SS', '000001.SZ']，然后在 handle_data 中使用 for stock in g.security: 循环，将后续逻辑中的 security 变量替换为循环变量 stock 即可。

Q: 为什么 MFI 指标有时候比 RSI 更有效？
A: RSI 只考虑价格变动，而 MFI 同时考虑了价格和成交量。在量价背离的行情中（例如价格上涨但成交量萎缩），MFI 能比 RSI 更早发出反转信号。

Q: 回测时没有交易是怎么回事？
A: 可能是因为 MFI 达到 20 以下或 80 以上的极端情况比较少见。你可以尝试放宽阈值，例如将买入阈值调整为 30，卖出阈值调整为 70，或者更换波动率更大的标的进行测试。