量化交易策略模板：Python/JavaScript示例
流量次数： 作者：小编 发布时间：2025-03-20 18:20:33
一、Python策略模板（均值回归策略）
1. 策略逻辑
信号生成：当价格偏离20日均线超过2倍标准差时反向开仓
止盈止损：固定比例止盈（3%）、动态跟踪止损（1.5倍ATR）
python
复制
import pandas as pd
import numpy as np
from bitget_api import BitgetAPIClient # 自定义API封装类
class MeanReversionStrategy:
def __init__(self, api_key, secret_key):
self.api = BitgetAPIClient(api_key, secret_key)
self.symbol = "BTCUSDT"
self.timeframe = "1d"
def get_historical_data(self):
"""获取历史K线数据"""
klines = self.api.get_klines(
symbol=self.symbol,
timeframe=self.timeframe,
limit=100
)
df = pd.DataFrame(klines, columns=['timestamp', 'open', 'high', 'low', 'close', 'volume'])
df['close'] = df['close'].astype(float)
return df
def calculate_indicators(self, df):
"""计算技术指标"""
df['ma20'] = df['close'].rolling(20).mean()
df['std20'] = df['close'].rolling(20).std()
df['upper_band'] = df['ma20'] + 2 * df['std20']
df['lower_band'] = df['ma20'] - 2 * df['std20']
return df
def generate_signal(self, df):
"""生成交易信号"""
latest = df.iloc[-1]
# 做空信号：价格突破上轨
if latest['close'] > latest['upper_band']:
return 'sell'
# 做多信号：价格跌破下轨
elif latest['close'] < latest['lower_band']:
return 'buy'
else:
return 'hold'
def execute_trade(self, signal):
"""执行交易逻辑"""
position = self.api.get_position(self.symbol)
if signal == 'buy' and position <= 0:
# 开多单，风险控制：仓位不超过总资金20%
self.api.place_order(
symbol=self.symbol,
side='buy',
amount=self.calculate_position_size(0.2),
stop_loss=self.calculate_stop_loss('buy')
)
elif signal == 'sell' and position >= 0:
# 开空单
self.api.place_order(...)
def run_strategy(self):
"""策略主循环"""
while True:
df = self.get_historical_data()
df = self.calculate_indicators(df)
signal = self.generate_signal(df)
self.execute_trade(signal)
time.sleep(60 * 60) # 每小时运行一次
if __name__ == "__main__":
strategy = MeanReversionStrategy(API_KEY, SECRET_KEY)
strategy.run_strategy()
二、JavaScript策略模板（动量突破策略）
1. 策略逻辑
信号生成：当价格突破过去N日最高价时做多，跌破最低价做空
仓位管理：凯利公式动态计算仓位比例
javascript
复制
const BitgetAPI = require('bitget-api-nodejs');
const { KellyCriterion } = require('./risk-management');
class MomentumStrategy {
constructor(apiKey, secretKey) {
this.api = new BitgetAPI(apiKey, secretKey);
this.symbol = 'ETHUSDT';
this.lookbackPeriod = 14;
}
async getHistoricalHighLow() {
const klines = await this.api.getKlines({
symbol: this.symbol,
timeframe: '1d',
limit: this.lookbackPeriod
});
const highs = klines.map(k => parseFloat(k[2]));
const lows = klines.map(k => parseFloat(k[3]));
return {
highestHigh: Math.max(...highs),
lowestLow: Math.min(...lows)
};
}
async checkBreakout() {
const currentPrice = await this.api.getTicker(this.symbol);
const { highestHigh, lowestLow } = await this.getHistoricalHighLow();
if (currentPrice > highestHigh) {
return 'long';
} else if (currentPrice < lowestLow) {
return 'short';
}
return 'no_signal';
}
async executeStrategy() {
const signal = await this.checkBreakout();
if (signal !== 'no_signal') {
const accountBalance = await this.api.getBalance();
const positionSize = KellyCriterion.calculate(
winRate: 0.55, // 历史胜率
payoffRatio: 1.5 // 盈亏比
) * accountBalance;
await this.api.placeOrder({
symbol: this.symbol,
side: signal,
amount: positionSize,
leverage: 3, // 3倍杠杆
stopLoss: signal === 'long' ? lowestLow : highestHigh
});
}
}
start() {
setInterval(() => this.executeStrategy(), 3600000); // 每小时执行
}
}
// 启动策略
const strategy = new MomentumStrategy(process.env.API_KEY, process.env.SECRET_KEY);
strategy.start();
三、策略核心模块分解
1. 数据获取层
python
复制
# Python示例：WebSocket实时数据订阅
import websockets
async def market_data_listener():
async with websockets.connect(BITGET_WS_URL) as ws:
await ws.send(subscribe_message)
while True:
data = await ws.recv()
process_market_data(data)
2. 风险控制模块
javascript
复制
// JavaScript凯利公式计算
class KellyCriterion {
static calculate(winRate, payoffRatio) {
const b = payoffRatio;
const p = winRate;
const q = 1 - p;
return (b * p - q) / b; // f* = (bp - q)/b
}
}
3. 订单执行优化
智能路由：根据深度选择Maker/Taker订单类型
冰山委托：大额订单拆分隐藏
防抖机制：防止信号频繁触发
四、策略回测与实盘切换
1. 回测框架结构（Python）
python
复制
class BacktestEngine:
def __init__(self, historical_data):
self.data = historical_data
self.equity_curve = []
def run_backtest(self, strategy):
for index, row in self.data.iterrows():
signal = strategy.generate_signal(row)
self.execute_signal(signal, row['close'])
def calculate_metrics(self):
# 计算夏普比率、最大回撤等
...
2. 实盘部署注意事项
使用独立服务器部署（推荐AWS/GCP）
启用日志监控（ELK Stack）
设置熔断机制（单日亏损>5%停止交易）
五、策略模板扩展方向
策略类型 特征 适用市场
套利策略 跨交易所价差捕捉 高流动性市场
做市策略 双边挂单赚取价差 低波动性市场
机器学习策略 LSTM/Transformer预测价格 大样本量市场
代码安全提示：
永远不要在代码中明文存储API密钥（使用环境变量或Vault服务）
对交易指令进行本地模拟测试后再接入实盘
定期审计策略逻辑防止参数过拟合