一、基于分钟级K线判断的期货交易策略

交易策略概述

这是一个基于分钟级K线判断的期货交易策略，主要特点包括：

• 交易标的: 白银期货 AG2604.SHF

• 交易周期: 1分钟K线

• 开仓逻辑: 根据前一分钟K线的涨跌方向决定开仓方向

• 平仓逻辑: 持仓满5分钟后自动平仓

• 交易方向: 阳线开多，阴线开空

策略代码

from datetime import datetime
import panda_quant
from panda_backtest.api.future_api import future_api_quotation
from panda_trading.trading_common.api.api import target_future_group_order, buy_open, sell_open, buy_close, sell_close
from panda_trading import SRLogger

# 全部合约列表（修改为白银2604）
SYMBOLS = [
    "AG2604.SHF"
]

def initialize(context):
    SRLogger.info("initialize")
    SRLogger.info('context.now: ' + str(context.now))
    # 获取当前分钟对应的下单合约索引
    now = datetime.now()
    print('时间', now)

def before_trading(context):
    SRLogger.info("before_trading")

def handle_data(context, data):
    SRLogger.info("handle_data")
    account = context.run_info.future_account

    # 获取当前分钟对应的下单合约索引（按分钟轮换合约，也可以直接用 SYMBOLS[0]）
    now = datetime.now()
    print('时间', now)
    idx = now.minute % len(SYMBOLS)
    symbol = SYMBOLS[idx]

    SRLogger.info(f"当前分钟: {now.minute}, 选择合约: {symbol}")

    # 使用上一交易日作为开始日期
    start_date = panda_quant.get_previous_trading_date(datetime.now().strftime('%Y%m%d'))
    end_date = datetime.now().strftime('%Y%m%d')

    # 打印入参
    SRLogger.info(f"调用参数: symbol_list={[symbol]}, start_date={start_date}, end_date={end_date}, period='1m'")

    quotation_df_1m = future_api_quotation(symbol_list=[symbol], start_date=start_date, end_date=end_date, period='1m')
    SRLogger.info(quotation_df_1m)

    # 数据不足则不交易
    if quotation_df_1m is None or len(quotation_df_1m) < 2:
        SRLogger.info(f"合约 {symbol} 的1分钟数据不足以判断K线阴阳")
        return

    # 按时间排序，确保最后一根是最新K线
    if "time" in quotation_df_1m.columns:
        quotation_df_1m = quotation_df_1m.sort_values(by=["date", "time"])
    else:
        quotation_df_1m = quotation_df_1m.sort_values(by=["date"])

    # 取前一根1分钟K线（倒数第二根）作为判断依据
    prev_bar = quotation_df_1m.iloc[-2]
    open_price = float(prev_bar["open"])
    close_price = float(prev_bar["close"])

    # 获取当前持仓（分别取多头和空头持仓数量）
    position = context.future_account_dict[account].positions[symbol]
    long_quantity = position.buy_quantity
    short_quantity = position.sell_quantity

    # ================= 开仓逻辑 =================
    # 条件：没有多头持仓且没有空头持仓时，根据前一分钟K线方向下单
    if long_quantity == 0 and short_quantity == 0:
        # 阳线：买入开多
        if close_price > open_price:
            buy_open(account, symbol, 1)
            SRLogger.info(f"前一分钟K线为阳线(开={open_price}, 收={close_price})，无多空持仓，买入开多 {symbol}")
            # 记录开仓时间与方向
            context.trade_state = getattr(context, "trade_state", {})
            context.trade_state[symbol] = {
                "side": "long",
                "open_minute": now
            }
        # 阴线：卖出开空
        elif close_price < open_price:
            sell_open(account, symbol, 1)
            SRLogger.info(f"前一分钟K线为阴线(开={open_price}, 收={close_price})，无多空持仓，卖出开空 {symbol}")
            # 记录开仓时间与方向
            context.trade_state = getattr(context, "trade_state", {})
            context.trade_state[symbol] = {
                "side": "short",
                "open_minute": now
            }
        else:
            # 十字线，不操作
            SRLogger.info(f"前一分钟K线为十字线(开={open_price}, 收={close_price})，无多空持仓，不开仓 {symbol}")
            return

    # ================= 平仓逻辑 =================
    else:
        # 已有持仓：持有满5分钟后全部平仓
        context.trade_state = getattr(context, "trade_state", {})
        state = context.trade_state.get(symbol, None)
        if state is None:
            # 若无记录，则不做时间平仓控制
            return

        open_minute = state.get("open_minute")
        side = state.get("side")

        # 使用分钟级时间差判断是否达到5分钟
        if isinstance(open_minute, datetime):
            held_minutes = (now - open_minute).total_seconds() / 60.0
        else:
            SRLogger.info(f"开仓时间记录异常，无法计算持仓时间: {open_minute}")
            return

        if held_minutes >= 5:
            if side == "long" and long_quantity > 0:
                # 平多
                sell_close(account, symbol, long_quantity)
                SRLogger.info(f"多头持仓已满5分钟，平仓 {symbol}，数量={long_quantity}")
            elif side == "short" and short_quantity > 0:
                # 平空
                buy_close(account, symbol, short_quantity)
                SRLogger.info(f"空头持仓已满5分钟，平仓 {symbol}，数量={short_quantity}")

            # 清除记录
            context.trade_state.pop(symbol, None)

def after_trading(context):
    SRLogger.info("after_trading")

策略工作流程

1. 初始化阶段

• 设置交易合约为 AG2604.SHF

• 记录当前时间

1. 数据获取

• 获取上一交易日到当前的1分钟K线数据

• 对数据进行时间排序处理

1. 开仓判断

• 无持仓时：

  • 前一分钟K线为阳线（收盘价 > 开盘价）：买入开多

  • 前一分钟K线为阴线（收盘价 < 开盘价）：卖出开空

  • 前一分钟K线为十字线（收盘价 = 开盘价）：不操作

• 记录开仓时间和方向

1. 平仓控制

• 持仓满5分钟后自动平仓

• 多头持仓到期时卖出平仓

• 空头持仓到期时买入平仓

• 平仓后清除交易记录

技术参数

• 交易标的: AG2604.SHF（白银期货）

• 数据周期: 1分钟K线

• 开仓条件: 前一分钟K线方向判断

• 持仓时间: 5分钟

• 仓位管理: 每次开仓1手

注意事项

1. 数据依赖: 策略依赖1分钟K线数据，数据不足时不会交易

2. 时间控制: 严格按照5分钟持仓时间执行平仓

3. 方向判断: 基于前一分钟K线的开收盘价判断涨跌

4. 持仓记录: 使用 context.trade_state 记录开仓信息，确保平仓逻辑正确执行

二、期货多品种仿真交易策略文档

策略概述

本策略是一个多品种期货仿真交易系统，采用基于均线的趋势跟踪策略，同时管理黄金、铜、螺纹钢、焦炭等多个期货品种的交易。每个品种独立执行相同的交易逻辑，通过价格与10分钟均线的交叉关系来决定开仓和平仓时机。

策略配置

交易品种

策略配置了以下期货品种：

注意：合约代码需要根据实际交易月份进行调整。

策略参数

• 默认交易手数：1手

• 均线周期：10根1分钟K线

• 交易时间周期：1分钟

• 数据获取区间：上一交易日至今

交易逻辑

初始化阶段

策略在初始化时会记录当前时间，并设置日志记录功能，确保交易过程的可追溯性。

数据获取

每个交易周期，策略会：

1. 获取上一交易日至今的1分钟行情数据

2. 验证数据完整性（至少10根K线）

3. 计算当前价格与10分钟均线的关系

开仓信号

当某个品种无持仓时：

• 做多开仓：当前价格 > 10分钟均线

• 做空开仓：当前价格 < 10分钟均线

• 观望：价格接近均线（不交易）

平仓信号

当某个品种有持仓时：

• 平多仓：价格上穿均线且持有多头仓位

• 平空仓：价格下穿均线且持有空头仓位

• 持仓：未触发平仓条件

风险控制机制

1. 数据验证：获取行情数据后检查数据量是否足够计算均线

2. 异常隔离：单个品种的交易异常不影响其他品种的正常运行

3. 持仓保护：平仓时自动计算可平仓手数，避免超平

4. 日志监控：完整的交易日志记录，便于策略调试和优化

代码实现

from datetime import datetime
import panda_quant
from panda_backtest.api.future_api import future_api_quotation
from panda_trading.trading_common.api.api import target_future_group_order, buy_open, sell_open, buy_close, sell_close
from panda_trading import SRLogger

# 全部合约列表：在这里配置要交易的黄金、铜、螺纹钢、焦炭等期货品种
# 合约代码需根据实际合约月份调整，这里仅示例 2604 合约
SYMBOLS = [
    "AU2604.SHF",  # 黄金
    "CU2603.SHF",  # 铜
    "RB2605.SHF",  # 螺纹钢
    "J2605.DCE",   # 焦炭（示例：大商所 J）
]

# 每次开/平仓默认手数
DEFAULT_VOLUME = 1

def initialize(context):
    SRLogger.info("initialize")
    SRLogger.info("context.now: " + str(context.now))
    now = datetime.now()
    print("初始化时间", now)

def before_trading(context):
    SRLogger.info("before_trading")

def _get_position_qty(context, account, symbol):
    """安全获取某个合约当前多头/空头持仓数量，防止 key 不存在时报错。"""
    acct = context.future_account_dict.get(account)
    if acct is None:
        return 0, 0
    positions = acct.positions
    pos = positions[symbol]
    if pos is None:
        return 0, 0
    buy_qty = pos.buy_quantity
    sell_qty = pos.sell_quantity
    return buy_qty, sell_qty

def handle_data(context, data):
    SRLogger.info("handle_data")
    account = context.run_info.future_account

    # 统一确定行情时间区间：上一交易日至今天
    today_str = datetime.now().strftime("%Y%m%d")
    start_date = panda_quant.get_previous_trading_date(today_str)
    end_date = today_str
    SRLogger.info(
        f"本轮处理时间: {datetime.now()}, start_date={start_date}, end_date={end_date}, period='1m'"
    )

    # 遍历黄金、铜、螺纹钢、焦炭等多个品种，逐一执行相同策略逻辑
    for symbol in SYMBOLS:
        try:
            SRLogger.info(f"开始处理合约: {symbol}")

            # 获取单品种 1 分钟行情
            SRLogger.info(
                f"调用参数: symbol_list={[symbol]}, start_date={start_date}, end_date={end_date}, period='1m'"
            )
            quotation_df = future_api_quotation(
                symbol_list=[symbol],
                start_date=start_date,
                end_date=end_date,
                period='1m',
            )
            SRLogger.info(f"{symbol} 行情数据: {quotation_df}")

            # 数据检查
            if quotation_df is None or len(quotation_df) < 10:
                SRLogger.info(
                    f"合约 {symbol} 的数据不足以计算均线 (len={0 if quotation_df is None else len(quotation_df)})"
                )
                continue

            close_prices = quotation_df["close"]

            # 计算最近 10 根 1 分钟 K 的均线（短周期均线）
            long_ma = close_prices.rolling(window=10).mean()
            ma_value = long_ma.iloc[-1]
            current_price = close_prices.iloc[-1]

            # 均线为 NaN 则跳过
            if ma_value != ma_value:  # NaN 判断
                SRLogger.info(f"合约 {symbol} 均线为 NaN，跳过")
                continue

            SRLogger.info(f"{symbol} 当前价: {current_price}, 10 均线: {ma_value}")

            # 获取当前持仓
            buy_quantity, sell_quantity = _get_position_qty(context, account, symbol)
            SRLogger.info(
                f"{symbol} 当前多头持仓: {buy_quantity}, 空头持仓: {sell_quantity}"
            )

            # 交易逻辑：与原策略相同，对每个品种独立执行
            if buy_quantity == 0 and sell_quantity == 0:
                # 无持仓：根据价格与均线关系开仓
                if current_price > ma_value:
                    buy_open(account, symbol, DEFAULT_VOLUME)
                    SRLogger.info(
                        f"[{symbol}] 当前价格大于10均线({current_price:.2f} > {ma_value:.2f})，买入开多 {DEFAULT_VOLUME} 手"
                    )
                elif current_price < ma_value:
                    sell_open(account, symbol, DEFAULT_VOLUME)
                    SRLogger.info(
                        f"[{symbol}] 当前价格小于10均线({current_price:.2f} < {ma_value:.2f})，卖出开空 {DEFAULT_VOLUME} 手"
                    )
                else:
                    SRLogger.info(f"[{symbol}] 当前价格接近均线，不开仓")
            else:
                # 有持仓：根据信号方向平仓
                if current_price > ma_value and sell_quantity > 0:
                    vol = min(DEFAULT_VOLUME, sell_quantity)
                    buy_close(account, symbol, vol)
                    SRLogger.info(
                        f"[{symbol}] 价格在10均线上方且存在空头，买入平空 {vol} 手"
                    )
                elif current_price < ma_value and buy_quantity > 0:
                    vol = min(DEFAULT_VOLUME, buy_quantity)
                    sell_close(account, symbol, vol)
                    SRLogger.info(
                        f"[{symbol}] 价格在10均线下方且存在多头，卖出平多 {vol} 手"
                    )
                else:
                    SRLogger.info(f"[{symbol}] 有持仓但未触发平仓条件")

        except Exception as e:
            # 单个品种异常不影响其他品种
            SRLogger.info(f"处理合约 {symbol} 时发生异常: {e}")

def after_trading(context):
    SRLogger.info("after_trading")

策略特点

优势

1. 多品种分散：同时交易多个品种，分散单一品种风险

2. 逻辑简单：基于经典均线交叉策略，易于理解和执行

3. 异常隔离：单个品种问题不影响整体策略运行

4. 完整日志：详细的交易记录便于监控和调试

局限性

1. 基础策略：未考虑更多技术指标或基本面因素

2. 固定参数：开平仓手数固定，未动态调整

3. 无资金管理：未实现复杂的资金管理策略

4. 滑点未考虑：实际交易中可能面临滑点影响

使用建议

1. 仿真测试：建议先在仿真环境充分测试策略表现

2. 参数优化：根据不同品种特性调整均线周期等参数

3. 风险控制：可考虑增加止损、仓位限制等风控措施

4. 实时监控：定期查看交易日志，关注策略运行状态

*点击查看视频教程：期货仿真录制教学视频

把中间代码复制到 Python输入那个节点框里面，在运行就可以了

本策略文档基于提供的JSON配置文件生成，仅供参考和学习使用,实际交易中请根据市场情况谨慎决策，注意控制风险。

如果需要帮助的话就可以添加小助理*