- 市场是动态的：过去有效的策略在未来可能会失效。
- 风险管理是核心：一个没有严格风险管理的策略，即使短期盈利，最终也可能因一次“黑天鹅”事件而爆仓。
- “圣杯”不存在：不存在一个能永远跑赢市场的策略。
- 法律与合规：程序化交易需要严格遵守交易所和券商的规定，否则可能导致账户被限制或处罚。
我将为您提供一个完整、结构清晰、功能齐全的期货程序化交易框架的Python源代码，这个框架包含了策略、风险管理、数据获取、交易执行和回测等核心模块，您可以用它作为基础，去实现和测试您自己的交易想法。
（图片来源网络，侵删）
核心思想与框架设计
一个好的程序化交易系统应该像搭积木一样,各个模块之间相互独立，又协同工作，我们将系统分为以下几个核心模块：
- 策略模块：交易的大脑，负责根据市场数据生成交易信号（买入、卖出、持有）。
- 数据模块：负责获取和处理历史数据或实时行情数据。
- 回测引擎：在历史数据上模拟策略的执行过程，评估策略的绩效。
- 风险管理模块：负责计算仓位、设置止损止盈等，是生存的关键。
- 交易执行模块：负责将交易指令发送到券商的交易接口。
我们将使用Python和几个流行的库来实现这个框架：
pandas: 用于数据处理。numpy: 用于科学计算。matplotlib: 用于结果可视化。vn.py(可选): 一个功能强大的开源量化交易平台，可以直接连接实盘，但我们这里先从基础的回测框架开始。
一个完整的双均线交叉策略回测框架
这是一个非常经典的趋势跟踪策略,逻辑很简单：
- 金叉：短期均线上穿长期均线，产生买入信号。
- 死叉：短期均线下穿长期均线，产生卖出信号。
1 项目结构
为了清晰,我们将代码分为几个文件：
（图片来源网络，侵删）
futures_strategy/
├── main.py # 主程序，运行回测
├── strategy.py # 策略模块
├── backtest.py # 回测引擎
├── risk_management.py # 风险管理模块
└── data_loader.py # 数据加载模块2 代码实现
data_loader.py - 数据加载模块
这个模块负责从CSV文件加载期货历史数据,假设你的CSV文件包含 datetime, open, high, low, close, volume 等列。
# data_loader.py
import pandas as pd
class DataLoader:
def __init__(self, file_path):
"""
初始化数据加载器
:param file_path: CSV文件路径
"""
self.file_path = file_path
self.data = None
def load_data(self):
"""
加载并预处理数据
:return: pandas DataFrame
"""
try:
self.data = pd.read_csv(self.file_path)
# 确保 'datetime' 列是 datetime 类型，并设置为索引
self.data['datetime'] = pd.to_datetime(self.data['datetime'])
self.data.set_index('datetime', inplace=True)
# 按时间排序
self.data.sort_index(inplace=True)
print(f"数据加载成功，共 {len(self.data)} 条记录。")
return self.data
except Exception as e:
print(f"数据加载失败: {e}")
return None
risk_management.py - 风险管理模块
这里实现一个简单的固定仓位管理,更高级的可以包括动态仓位、止损止盈等。
# risk_management.py
class RiskManager:
def __init__(self, initial_capital, risk_per_trade=0.02):
"""
初始化风险管理器
:param initial_capital: 初始资金
:param risk_per_trade: 每笔交易的风险比例 (e.g., 0.02 表示2%)
"""
self.initial_capital = initial_capital
self.capital = initial_capital
self.risk_per_trade = risk_per_trade
def calculate_position_size(self, entry_price, stop_loss_price):
"""
根据风险计算仓位大小
:param entry_price: 入场价格
:param stop_loss_price: 止损价格
:return: 应该买入/卖出的手数
"""
if entry_price <= stop_loss_price:
print("错误：入场价格必须大于止损价格（多头）或小于（空头）。")
return 0
# 计算每手合约的风险金额 (这里简化处理，实际需要知道合约乘数)
# 假设每手合约价值为 entry_price * 10 (以螺纹钢为例)
contract_value = entry_price * 10
risk_per_unit = abs(entry_price - stop_loss_price) * 10
if risk_per_unit == 0:
return 0
# 计算可以承受的总风险金额
total_risk_amount = self.capital * self.risk_per_trade
# 计算手数
position_size = total_risk_amount / risk_per_unit
# 向下取整，确保不会超风险
return int(position_size)
def update_capital(self, trade_result):
"""
根据交易结果更新资金
:param trade_result: 交易盈亏金额
"""
self.capital += trade_result
print(f"资金更新为: {self.capital:.2f}")
strategy.py - 策略模块
这是策略的核心,我们在这里定义双均线交叉逻辑。
# strategy.py
import pandas as pd
class MovingAverageCrossStrategy:
def __init__(self, short_window=5, long_window=20):
"""
初始化双均线策略
:param short_window: 短期均线窗口
:param long_window: 长期均线窗口
"""
self.short_window = short_window
self.long_window = long_window
self.signals = pd.DataFrame(index=data.index)
self.signals['signal'] = 0.0
def generate_signals(self, data):
"""
根据价格数据生成交易信号
:param data: 包含 'close' 价格的 DataFrame
"""
# 计算移动平均线
self.signals['short_mavg'] = data['close'].rolling(window=self.short_window, min_periods=1).mean()
self.signals['long_mavg'] = data['close'].rolling(window=self.long_window, min_periods=1).mean()
# 生成信号
# 当短期均线上穿长期均线时，生成1（买入）信号
self.signals['signal'][self.short_window:] = np.where(
self.signals['short_mavg'][self.short_window:] > self.signals['long_mavg'][self.short_window:], 1.0, 0.0)
# 计算持仓变化：当信号从0变为1时，为1；从1变为0时，为-1
self.signals['positions'] = self.signals['signal'].diff()
return self.signals
backtest.py - 回测引擎
这是模拟交易执行的核心模块,它协调策略、数据和风险管理。
（图片来源网络，侵删）
# backtest.py
import pandas as pd
import matplotlib.pyplot as plt
class BacktestEngine:
def __init__(self, data, strategy, risk_manager):
"""
初始化回测引擎
:param data: 历史数据
:param strategy: 策略对象
:param risk_manager: 风险管理对象
"""
self.data = data
self.strategy = strategy
self.risk_manager = risk_manager
self.portfolio = pd.DataFrame(index=data.index)
self.portfolio['holdings'] = 0.0 # 持仓数量
self.portfolio['cash'] = self.risk_manager.capital # 现金
self.portfolio['total'] = self.risk_manager.capital # 总资产
self.portfolio['returns'] = 0.0 # 收益率
def run(self):
"""
运行回测
"""
print("开始回测...")
# 生成交易信号
signals = self.strategy.generate_signals(self.data)
for i in range(1, len(self.data)):
current_date = self.data.index[i]
previous_date = self.data.index[i-1]
# 当前信号
current_signal = signals.loc[current_date, 'positions']
if current_signal > 0: # 买入信号
print(f"{current_date}: 买入信号")
entry_price = self.data.loc[current_date, 'close']
# 简化止损：入场价格的2%
stop_loss_price = entry_price * 0.98
position_size = self.risk_manager.calculate_position_size(entry_price, stop_loss_price)
if position_size > 0:
cost = entry_price * position_size * 10 # 假设合约乘数为10
if cost <= self.portfolio.loc[previous_date, 'cash']:
self.portfolio.loc[current_date, 'holdings'] = position_size
self.portfolio.loc[current_date, 'cash'] = self.portfolio.loc[previous_date, 'cash'] - cost
print(f" 以 {entry_price} 买入 {position_size} 手")
elif current_signal < 0: # 卖出信号
print(f"{current_date}: 卖出信号")
position_size = self.portfolio.loc[previous_date, 'holdings']
if position_size > 0:
exit_price = self.data.loc[current_date, 'close']
revenue = exit_price * position_size * 10
self.portfolio.loc[current_date, 'holdings'] = 0
self.portfolio.loc[current_date, 'cash'] = self.portfolio.loc[previous_date, 'cash'] + revenue
profit = revenue - (self.portfolio.loc[previous_date, 'total'] - self.portfolio.loc[previous_date, 'cash'])
print(f" 以 {exit_price} 卖出 {position_size} 手, 盈利: {profit:.2f}")
self.risk_manager.update_capital(profit)
# 更新总资产
self.portfolio.loc[current_date, 'total'] = (
self.portfolio.loc[current_date, 'cash'] +
self.portfolio.loc[current_date, 'holdings'] * self.data.loc[current_date, 'close'] * 10
)
# 计算收益率
self.portfolio.loc[current_date, 'returns'] = (
self.portfolio.loc[current_date, 'total'] / self.portfolio.loc[previous_date, 'total'] - 1
)
print("回测结束。")
return self.portfolio
def plot_results(self):
"""
绘制回测结果
"""
fig, (ax1, ax2) = plt.subplots(2, 1, figsize=(12, 8), sharex=True)
# 绘制价格和移动平均线
ax1.plot(self.data.index, self.data['close'], label='Price', color='k', alpha=0.7)
ax1.plot(self.strategy.signals.index, self.strategy.signals['short_mavg'], label='Short MA')
ax1.plot(self.strategy.signals.index, self.strategy.signals['long_mavg'], label='Long MA')
# 绘制买入卖出点
ax1.plot(self.strategy.signals.loc[self.strategy.signals['positions'] == 1.0].index,
self.strategy.signals['short_mavg'][self.strategy.signals['positions'] == 1.0],
'^', markersize=10, color='g', lw=0)
ax1.plot(self.strategy.signals.loc[self.strategy.signals['positions'] == -1.0].index,
self.strategy.signals['short_mavg'][self.strategy.signals['positions'] == -1.0],
'v', markersize=10, color='r', lw=0)
ax1.set_title('Price and Moving Averages')
ax1.legend()
ax1.grid(True)
# 绘制资产曲线
ax2.plot(self.portfolio.index, self.portfolio['total'], label='Portfolio Total', color='b')
ax2.set_title('Portfolio Value Over Time')
ax2.legend()
ax2.grid(True)
plt.tight_layout()
plt.show()
main.py - 主程序
将所有模块串联起来,运行回测。
# main.py
import numpy as np
from data_loader import DataLoader
from strategy import MovingAverageCrossStrategy
from risk_management import RiskManager
from backtest import BacktestEngine
if __name__ == "__main__":
# --- 1. 参数设置 ---
# 假设你有一份名为 'rb_data.csv' 的螺纹钢期货数据
data_file = 'rb_data.csv'
initial_capital = 100000 # 初始资金10万
short_window = 5 # 短期均线
long_window = 20 # 长期均线
# --- 2. 加载数据 ---
loader = DataLoader(data_file)
data = loader.load_data()
if data is None:
exit()
# --- 3. 初始化模块 ---
strategy = MovingAverageCrossStrategy(short_window=short_window, long_window=long_window)
risk_manager = RiskManager(initial_capital=initial_capital)
backtest_engine = BacktestEngine(data, strategy, risk_manager)
# --- 4. 运行回测 ---
portfolio = backtest_engine.run()
# --- 5. 输出结果和绘图 ---
print("\n--- 回测绩效摘要 ---")
final_value = portfolio['total'][-1]
total_return = (final_value - initial_capital) / initial_capital * 100
print(f"初始资金: {initial_capital:,.2f}")
print(f"最终资金: {final_value:,.2f}")
print(f"总收益率: {total_return:.2f}%")
# 绘制图表
backtest_engine.plot_results()
如何使用这个框架
- 准备数据：创建一个CSV文件（
rb_data.csv如下（可以用从Tushare、Wind等获取的数据）：datetime,open,high,low,close,volume 2025-01-03,4000,4020,3990,4010,100000 2025-01-04,4010,4030,4005,4025,110000 ... (更多数据)
- 安装依赖：
pip install pandas numpy matplotlib
- 运行主程序：
python main.py
你将看到回测过程的日志输出,最后会显示一个包含价格走势和资产曲线的图表。
从回测到实盘的下一步
这个框架是回测,要用于实盘，你需要考虑：
-
交易接口：
- 券商提供的API：国内主流券商（如华泰、中信、国泰君安等）都提供了程序化交易接口（CTP、极速柜台等），你需要申请并接入他们的API。
- 第三方平台：
vn.py是一个基于Python的开源量化交易平台，它封装了多家券商的接口，提供了强大的实盘和回测功能，是进入实盘的绝佳选择。
-
实盘与回测的差异：
- 滑点：回测中假设能按理想价格成交，但实盘中会有滑点，你的策略必须能承受一定的滑点。
- 手续费：回测中必须精确计算交易所和券商收取的手续费、印花税等。
- 流动性：在流动性差的品种或时段，大单交易会对价格产生冲击。
- 数据延迟：实盘数据有网络延迟，回测用的是“上帝视角”的完美数据。
-
持续监控与迭代：
- 实盘运行后,需要7x24小时监控策略的运行状态。
- 定期评估策略表现,当市场环境变化导致策略失效时，要及时停止或迭代策略。
免责声明：以上代码和框架仅供学习和研究使用，不构成任何投资建议，金融市场有风险，程序化交易风险极高，请在充分了解并控制风险的前提下进行实盘操作。
标签： 期货程序化交易开源代码 期货量化交易策略源码下载 期货交易系统源代码获取渠道
版权声明：除非特别标注，否则均为本站原创文章，转载时请以链接形式注明文章出处。
