第192页 · 预计6页
本附录提供了壹信量化套利策略的核心Python代码片段，涵盖期现套利、资金费率套利、跨所套利、配对统计套利、网格套利五大核心策略。所有代码均为简化版，重点展示策略逻辑和API调用方式，适合作为策略开发的参考模板。
requests, pandas, numpy, ccxt, python-dotenvpip install requests pandas numpy ccxt python-dotenv壹信量化平台提供两类API：
https://api.a-sig.com/v1X-API-Key）https://api.intoquant.com/v1注意：本代码中的API端点和参数为示例，实际使用时请参考壹信量化平台的最新API文档。你可以在 a-sig.com 和 intoquant.com 的开发者中心获取完整的API文档和密钥。
每个策略代码包含以下模块：
所有代码都有详细的中文注释，便于理解和修改。
本代码仅供学习和参考，不构成投资建议。实盘交易前请充分回测和模拟盘验证，并自行承担交易风险。
在具体策略之前，先提供一个通用的工具模块，包含API请求封装、指标计算、日志记录等通用功能，后续策略都会引用这个模块。
"""
文件：yixin_common.py
功能：壹信量化通用工具模块
"""
import time
import hmac
import hashlib
import requests
import pandas as pd
import numpy as np
from datetime import datetime
from dotenv import load_dotenv
import os
# 加载环境变量（API密钥等敏感信息放在.env文件中，不要硬编码在代码里）
load_dotenv()
# ==================== 配置 ====================
A_SIG_BASE_URL = "https://api.a-sig.com/v1" # 壹信数据API基础URL
INTOQUANT_BASE_URL = "https://api.intoquant.com/v1" # 壹信交易API基础URL
A_SIG_API_KEY = os.getenv("A_SIG_API_KEY", "") # 壹信数据API密钥
INTOQUANT_API_KEY = os.getenv("INTOQUANT_API_KEY", "") # 壹信交易API Key
INTOQUANT_API_SECRET = os.getenv("INTOQUANT_API_SECRET", "") # 壹信交易API Secret
# ==================== 日志工具 ====================
def log(level, message):
"""
简单的日志函数，打印带时间戳和级别的日志
实际项目中建议使用logging模块，写入文件并做日志轮转
"""
timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
print(f"[{timestamp}] [{level}] {message}")
# ==================== API请求工具 ====================
def a_sig_request(endpoint, params=None):
"""
向壹信数据API（a-sig.com）发送GET请求
:param endpoint: API端点，如 "/market/klines"
:param params: 请求参数字典
:return: 返回JSON数据
"""
url = f"{A_SIG_BASE_URL}{endpoint}"
headers = {
"X-API-Key": A_SIG_API_KEY,
"Content-Type": "application/json"
}
try:
response = requests.get(url, headers=headers, params=params, timeout=10)
response.raise_for_status() # 抛出HTTP错误
return response.json()
except requests.exceptions.RequestException as e:
log("ERROR", f"a-sig API请求失败: {endpoint}, 错误: {e}")
return None
def intoquant_sign(params, secret):
"""
生成壹信交易API的HMAC-SHA256签名
:param params: 请求参数字典（包含timestamp）
:param secret: API Secret
:return: 签名字符串
"""
# 将参数按key排序，拼接成查询字符串
sorted_params = sorted(params.items())
query_string = "&".join([f"{k}={v}" for k, v in sorted_params])
# HMAC-SHA256签名
signature = hmac.new(
secret.encode(),
query_string.encode(),
hashlib.sha256
).hexdigest()
return signature
def intoquant_request(method, endpoint, params=None):
"""
向壹信交易API（intoquant.com）发送签名请求
:param method: "GET" 或 "POST"
:param endpoint: API端点，如 "/order/create"
:param params: 请求参数字典
:return: 返回JSON数据
"""
url = f"{INTOQUANT_BASE_URL}{endpoint}"
timestamp = int(time.time() * 1000) # 毫秒时间戳
# 构造参数字典，加入时间戳
if params is None:
params = {}
params["timestamp"] = timestamp
# 生成签名
signature = intoquant_sign(params, INTOQUANT_API_SECRET)
params["sign"] = signature
headers = {
"X-API-Key": INTOQUANT_API_KEY,
"Content-Type": "application/json"
}
try:
if method.upper() == "GET":
response = requests.get(url, headers=headers, params=params, timeout=10)
else:
response = requests.post(url, headers=headers, json=params, timeout=10)
response.raise_for_status()
return response.json()
except requests.exceptions.RequestException as e:
log("ERROR", f"intoquant API请求失败: {method} {endpoint}, 错误: {e}")
return None
# ==================== 技术指标计算工具 ====================
def calculate_sma(series, window):
"""
计算简单移动平均线（SMA）
:param series: 价格序列（pandas.Series）
:param window: 窗口期
:return: SMA序列
"""
return series.rolling(window=window).mean()
def calculate_std(series, window):
"""
计算滚动标准差
:param series: 价格序列
:param window: 窗口期
:return: 标准差序列
"""
return series.rolling(window=window).std()
def calculate_zscore(series, window):
"""
计算滚动Z-Score
:param series: 数据序列（如价差）
:param window: 滚动窗口
:return: Z-Score序列
"""
mean = series.rolling(window=window).mean()
std = series.rolling(window=window).std()
zscore = (series - mean) / std
return zscore
def calculate_atr(high, low, close, window=14):
"""
计算平均真实波幅（ATR）
:param high: 最高价序列
:param low: 最低价序列
:param close: 收盘价序列
:param window: 窗口期
:return: ATR序列
"""
# 计算真实波幅TR
tr1 = high - low
tr2 = abs(high - close.shift(1))
tr3 = abs(low - close.shift(1))
tr = pd.concat([tr1, tr2, tr3], axis=1).max(axis=1)
# 计算ATR（简单移动平均）
atr = tr.rolling(window=window).mean()
return atr
def calculate_rsi(close, window=14):
"""
计算相对强弱指数（RSI）
:param close: 收盘价序列
:param window: 窗口期
:return: RSI序列
"""
delta = close.diff()
gain = delta.where(delta > 0, 0)
loss = (-delta).where(delta < 0, 0)
avg_gain = gain.rolling(window=window).mean()
avg_loss = loss.rolling(window=window).mean()
rs = avg_gain / avg_loss
rsi = 100 - (100 / (1 + rs))
return rsi
# ==================== 风控工具 ====================
def check_drawdown(current_value, peak_value, max_drawdown_limit):
"""
检查当前回撤是否超过限制
:param current_value: 当前资产总值
:param peak_value: 历史最高资产总值
:param max_drawdown_limit: 最大回撤限制（如0.1表示10%）
:return: (是否超限, 当前回撤)
"""
if peak_value == 0:
return False, 0
drawdown = (peak_value - current_value) / peak_value
is_exceeded = drawdown > max_drawdown_limit
return is_exceeded, drawdown
def check_daily_loss(today_start_value, current_value, daily_loss_limit):
"""
检查当日亏损是否超过限制
:param today_start_value: 今日初始资产
:param current_value: 当前资产
:param daily_loss_limit: 单日亏损限制（如0.03表示3%）
:return: (是否超限, 当日亏损率)
"""
if today_start_value == 0:
return False, 0
daily_loss = (today_start_value - current_value) / today_start_value
is_exceeded = daily_loss > daily_loss_limit
return is_exceeded, daily_loss
# ==================== 数据获取工具 ====================
def get_klines(symbol, interval, limit=100):
"""
从壹信数据API获取K线数据
:param symbol: 交易对，如 "BTC/USDT"
:param interval: K线周期，如 "1m", "5m", "1h", "1d"
:param limit: 获取数量
:return: DataFrame，包含 open, high, low, close, volume
"""
params = {
"symbol": symbol,
"interval": interval,
"limit": limit
}
data = a_sig_request("/market/klines", params)
if data and "data" in data:
df = pd.DataFrame(data["data"])
# 确保列名和数据类型正确
df["timestamp"] = pd.to_datetime(df["timestamp"], unit="ms")
for col in ["open", "high", "low", "close", "volume"]:
df[col] = df[col].astype(float)
df.set_index("timestamp", inplace=True)
return df
else:
log("ERROR", f"获取K线数据失败: {symbol} {interval}")
return None
def get_funding_rate(symbol, exchange="binance"):
"""
获取当前资金费率
:param symbol: 交易对，如 "BTC/USDT"
:param exchange: 交易所，如 "binance", "okx"
:return: 当前资金费率（如0.0001表示0.01%）
"""
params = {
"symbol": symbol,
"exchange": exchange
}
data = a_sig_request("/futures/funding-rate", params)
if data and "data" in data:
return float(data["data"]["fundingRate"])
else:
log("ERROR", f"获取资金费率失败: {symbol} {exchange}")
return None
def get_spread(symbol, exchange_a, exchange_b):
"""
获取两个交易所之间的价差
:param symbol: 交易对
:param exchange_a: 交易所A
:param exchange_b: 交易所B
:return: 价差（绝对）和价差率
"""
params = {
"symbol": symbol,
"exchange_a": exchange_a,
"exchange_b": exchange_b
}
data = a_sig_request("/arbitrage/spread", params)
if data and "data" in data:
return float(data["data"]["spread"]), float(data["data"]["spreadRate"])
else:
log("ERROR", f"获取价差失败: {symbol} {exchange_a}-{exchange_b}")
return None, None
期现套利是最基础、最稳健的套利策略。当期货价格与现货价格的基差扩大到一定程度时，做多现货+做空期货，等基差收敛后平仓获利。
"""
文件：strategy_cash_futures_arbitrage.py
功能：BTC期现套利策略（简化版）
策略逻辑：
1. 实时计算BTC现货和当季期货的基差率
2. 当基差率 > 开仓阈值（如1.5%）时，做多现货+做空期货
3. 当基差率 < 平仓阈值（如0.3%）时，平仓获利
4. 严格的风控：最大回撤、单日亏损、仓位限制
"""
import time
import pandas as pd
from yixin_common import (
log, get_klines, intoquant_request,
check_drawdown, check_daily_loss, calculate_sma
)
# ==================== 策略参数 ====================
SYMBOL = "BTC/USDT" # 交易对
SPOT_EXCHANGE = "binance" # 现货交易所
FUTURES_EXCHANGE = "binance" # 期货交易所
FUTURES_TYPE = "quarterly" # 期货类型：quarterly(当季), perpetual(永续)
BASIS_OPEN_THRESHOLD = 0.015 # 基差率开仓阈值：1.5%
BASIS_CLOSE_THRESHOLD = 0.003 # 基差率平仓阈值：0.3%
POSITION_SIZE_RATIO = 0.8 # 仓位比例：使用80%的可用资金
MAX_DRAWDOWN_LIMIT = 0.08 # 最大回撤限制：8%
DAILY_LOSS_LIMIT = 0.03 # 单日亏损限制：3%
CHECK_INTERVAL = 60 # 检查间隔：60秒
# ==================== 策略状态 ====================
class StrategyState:
def __init__(self):
self.position = 0 # 当前持仓状态：0=空仓, 1=持有期现头寸
self.spot_qty = 0 # 现货持仓数量
self.futures_qty = 0 # 期货持仓数量
self.entry_basis = 0 # 开仓时的基差率
self.peak_value = 0 # 历史最高资产总值
self.today_start_value = 0 # 今日初始资产
self.last_check_date = None # 上次检查日期（用于重置今日初始值）
state = StrategyState()
# ==================== 核心函数 ====================
def get_account_value():
"""
获取账户总资产（现货+期货权益合计）
"""
result = intoquant_request("GET", "/account/total-value")
if result and result.get("code") == 0:
return float(result["data"]["totalValue"])
else:
log("ERROR", "获取账户总资产失败")
return None
def get_current_basis():
"""
获取当前基差率
基差率 = (期货价格 - 现货价格) / 现货价格
"""
# 获取现货价格
spot_data = intoquant_request("GET", "/market/ticker",
{"symbol": SYMBOL, "exchange": SPOT_EXCHANGE, "type": "spot"})
# 获取期货价格
futures_data = intoquant_request("GET", "/market/ticker",
{"symbol": SYMBOL, "exchange": FUTURES_EXCHANGE, "type": "futures", "futuresType": FUTURES_TYPE})
if spot_data and futures_data and spot_data.get("code") == 0 and futures_data.get("code") == 0:
spot_price = float(spot_data["data"]["last"])
futures_price = float(futures_data["data"]["last"])
basis_rate = (futures_price - spot_price) / spot_price
return basis_rate, spot_price, futures_price
else:
log("ERROR", "获取当前价格失败")
return None, None, None
def open_position(basis_rate, spot_price, futures_price, account_value):
"""
开仓：做多现货 + 做空期货
"""
log("INFO", f"触发开仓信号，当前基差率: {basis_rate:.4%}，开始开仓...")
# 计算仓位金额
position_value = account_value * POSITION_SIZE_RATIO
# 现货买入数量（用一半资金买现货）
spot_budget = position_value / 2
spot_qty = spot_budget / spot_price
# 期货做空数量（跟现货数量对应，1倍杠杆）
futures_qty = spot_qty
# 买入现货
spot_order = intoquant_request("POST", "/order/create", {
"exchange": SPOT_EXCHANGE,
"symbol": SYMBOL,
"type": "spot",
"side": "buy",
"orderType": "market", # 市价单，实际建议用限价单降低滑点
"quantity": round(spot_qty, 6)
})
# 做空期货
futures_order = intoquant_request("POST", "/order/create", {
"exchange": FUTURES_EXCHANGE,
"symbol": SYMBOL,
"type": "futures",
"futuresType": FUTURES_TYPE,
"side": "sell",
"orderType": "market",
"quantity": round(futures_qty, 6),
"leverage": 1 # 1倍杠杆
})
if spot_order and futures_order and spot_order.get("code") == 0 and futures_order.get("code") == 0:
log("INFO", f"开仓成功！现货买入: {spot_qty:.6f} BTC，期货做空: {futures_qty:.6f} BTC")
state.position = 1
state.spot_qty = spot_qty
state.futures_qty = futures_qty
state.entry_basis = basis_rate
return True
else:
log("ERROR", "开仓失败，请检查订单状态")
return False
def close_position(basis_rate):
"""
平仓：卖出现货 + 平掉期货空头
"""
log("INFO", f"触发平仓信号，当前基差率: {basis_rate:.4%}，开仓基差率: {state.entry_basis:.4%}，开始平仓...")
# 卖出现货
spot_order = intoquant_request("POST", "/order/create", {
"exchange": SPOT_EXCHANGE,
"symbol": SYMBOL,
"type": "spot",
"side": "sell",
"orderType": "market",
"quantity": round(state.spot_qty, 6)
})
# 平期货空头（买入平仓）
futures_order = intoquant_request("POST", "/order/create", {
"exchange": FUTURES_EXCHANGE,
"symbol": SYMBOL,
"type": "futures",
"futuresType": FUTURES_TYPE,
"side": "buy",
"orderType": "market",
"quantity": round(state.futures_qty, 6),
"reduceOnly": True # 只减仓
})
if spot_order and futures_order and spot_order.get("code") == 0 and futures_order.get("code") == 0:
profit = (state.entry_basis - basis_rate) * state.spot_qty * 100 # 粗略估算收益
log("INFO", f"平仓成功！本次套利预估收益: {profit:.2f} USDT")
state.position = 0
state.spot_qty = 0
state.futures_qty = 0
state.entry_basis = 0
return True
else:
log("ERROR", "平仓失败，请检查订单状态")
return False
def risk_check(account_value):
"""
风控检查：最大回撤、单日亏损
返回True表示可以继续交易，False表示需要暂停交易
"""
# 更新历史最高值
if account_value > state.peak_value:
state.peak_value = account_value
# 检查最大回撤
dd_exceeded, drawdown = check_drawdown(account_value, state.peak_value, MAX_DRAWDOWN_LIMIT)
if dd_exceeded:
log("WARNING", f"最大回撤超限！当前回撤: {drawdown:.2%}，限制: {MAX_DRAWDOWN_LIMIT:.2%}，暂停开仓")
if state.position == 1:
log("WARNING", "当前有持仓，触发回撤止损，强制平仓")
basis_rate, _, _ = get_current_basis()
if basis_rate is not None:
close_position(basis_rate)
return False
# 检查单日亏损
today = time.strftime("%Y-%m-%d")
if state.last_check_date != today:
state.today_start_value = account_value
state.last_check_date = today
loss_exceeded, daily_loss = check_daily_loss(state.today_start_value, account_value, DAILY_LOSS_LIMIT)
if loss_exceeded:
log("WARNING", f"单日亏损超限！当日亏损: {daily_loss:.2%}，限制: {DAILY_LOSS_LIMIT:.2%}，今日暂停交易")
if state.position == 1:
log("WARNING", "当前有持仓，触发单日亏损止损，强制平仓")
basis_rate, _, _ = get_current_basis()
if basis_rate is not None:
close_position(basis_rate)
return False
return True
# ==================== 主循环 ====================
def main():
log("INFO", "=" * 50)
log("INFO", "BTC期现套利策略启动")
log("INFO", f"开仓基差阈值: {BASIS_OPEN_THRESHOLD:.2%}")
log("INFO", f"平仓基差阈值: {BASIS_CLOSE_THRESHOLD:.2%}")
log("INFO", "=" * 50)
while True:
try:
# 1. 获取账户资产
account_value = get_account_value()
if account_value is None:
time.sleep(CHECK_INTERVAL)
continue
# 2. 风控检查
if not risk_check(account_value):
time.sleep(CHECK_INTERVAL)
continue
# 3. 获取当前基差
basis_rate, spot_price, futures_price = get_current_basis()
if basis_rate is None:
time.sleep(CHECK_INTERVAL)
continue
log("INFO", f"当前基差率: {basis_rate:.4%}，持仓状态: {state.position}")
# 4. 信号判断和交易执行
if state.position == 0:
# 空仓状态：检查开仓信号
if basis_rate > BASIS_OPEN_THRESHOLD:
open_position(basis_rate, spot_price, futures_price, account_value)
else:
# 持仓状态：检查平仓信号
if basis_rate < BASIS_CLOSE_THRESHOLD:
close_position(basis_rate)
# 5. 等待下一次检查
time.sleep(CHECK_INTERVAL)
except Exception as e:
log("ERROR", f"主循环异常: {e}")
time.sleep(CHECK_INTERVAL)
if __name__ == "__main__":
main()
期现套利策略使用说明：
参数调整：根据市场环境调整BASIS_OPEN_THRESHOLD和BASIS_CLOSE_THRESHOLD。牛市中基差通常较高，可以把开仓阈值设高一些（如2%）；熊市中基差较低甚至为负，可以设低一些（如0.8%）。建议用a-sig.com的基差历史数据统计合理的阈值。
下单方式：示例中用的是市价单，实盘中建议改用限价单（orderType: "limit"，指定price参数），可以显著降低滑点。可以配合"限价单+超时撤销+市价单兜底"的智能下单算法。
期货类型：示例中用的是当季期货（quarterly），你也可以用永续合约（perpetual）。用永续合约时，需要考虑资金费率的影响——如果资金费率为正，做空期货可以额外收取资金费，增加收益；如果为负，则需要支付资金费，减少收益。
基差计算优化：示例中用最新成交价计算基差，实盘中建议用订单簿的中间价（(买一价+卖一价)/2）计算基差，更准确地反映可成交的基差。可以调用/market/orderbook接口获取订单簿数据。
对冲比例：示例中用1:1的对冲比例（现货数量=期货数量），这是简化处理。实际上，因为期货有杠杆和保证金，更精确的对冲需要考虑delta对冲、保证金成本等。对于1倍杠杆的期现套利，1:1基本够用。
资金费率套利利用永续合约的资金费率机制，当资金费率为正时，做空永续合约+做多现货，收取资金费；当资金费率为负时反向操作。
"""
文件：strategy_funding_rate_arbitrage.py
功能：资金费率套利策略（简化版）
策略逻辑：
1. 实时监控永续合约的资金费率
2. 当资金费率 > 正费率阈值（如0.05%每8小时）时，做多现货+做空永续，收取资金费
3. 当资金费率 < 负费率阈值（如-0.05%每8小时）时，做空现货+做多永续，收取资金费
4. 当资金费率回归中性区间时，平仓
5. 同时监控价格波动风险，设置止损
"""
import time
from yixin_common import (
log, get_funding_rate, intoquant_request,
check_drawdown, check_daily_loss
)
# ==================== 策略参数 ====================
SYMBOL = "BTC/USDT"
EXCHANGE = "binance"
FUNDING_LONG_THRESHOLD = 0.0005 # 正费率开仓阈值：0.05%（每8小时），做空永续+做多现货
FUNDING_SHORT_THRESHOLD = -0.0005 # 负费率开仓阈值：-0.05%（每8小时），做多永续+做空现货
FUNDING_CLOSE_THRESHOLD = 0.0001 # 费率平仓阈值：绝对值<0.01%时平仓
POSITION_SIZE_RATIO = 0.7 # 仓位比例
MAX_DRAWDOWN_LIMIT = 0.10 # 最大回撤限制：10%
DAILY_LOSS_LIMIT = 0.04 # 单日亏损限制：4%
PRICE_STOP_LOSS = 0.05 # 价格波动止损：对冲头寸的价格偏离超过5%时止损
CHECK_INTERVAL = 300 # 检查间隔：5分钟（资金费率变化慢，不需要太频繁）
# ==================== 策略状态 ====================
class StrategyState:
def __init__(self):
self.position = 0 # 0=空仓, 1=正费率套利(多现货+空永续), -1=负费率套利(空现货+多永续)
self.spot_qty = 0
self.futures_qty = 0
self.entry_funding = 0 # 开仓时的资金费率
self.entry_spot_price = 0 # 开仓时的现货价格
self.peak_value = 0
self.today_start_value = 0
self.last_check_date = None
state = StrategyState()
# ==================== 核心函数 ====================
def get_account_value():
result = intoquant_request("GET", "/account/total-value")
if result and result.get("code") == 0:
return float(result["data"]["totalValue"])
return None
def get_current_prices():
"""获取当前现货和永续价格"""
spot = intoquant_request("GET", "/market/ticker",
{"symbol": SYMBOL, "exchange": EXCHANGE, "type": "spot"})
futures = intoquant_request("GET", "/market/ticker",
{"symbol": SYMBOL, "exchange": EXCHANGE, "type": "futures", "futuresType": "perpetual"})
if spot and futures and spot.get("code") == 0 and futures.get("code") == 0:
return float(spot["data"]["last"]), float(futures["data"]["last"])
return None, None
def open_position(funding_rate, direction, account_value):
"""
开仓
direction=1: 正费率套利 -> 做多现货 + 做空永续
direction=-1: 负费率套利 -> 做空现货(或借币卖空) + 做多永续
"""
log("INFO", f"触发开仓信号，资金费率: {funding_rate:.4%}，方向: {'正费率套利' if direction==1 else '负费率套利'}")
position_value = account_value * POSITION_SIZE_RATIO
spot_price, futures_price = get_current_prices()
if spot_price is None:
return False
qty = (position_value / 2) / spot_price # 现货和期货各用一半资金
if direction == 1:
# 做多现货
spot_order = intoquant_request("POST", "/order/create", {
"exchange": EXCHANGE, "symbol": SYMBOL, "type": "spot",
"side": "buy", "orderType": "market", "quantity": round(qty, 6)
})
# 做空永续
futures_order = intoquant_request("POST", "/order/create", {
"exchange": EXCHANGE, "symbol": SYMBOL, "type": "futures",
"futuresType": "perpetual", "side": "sell", "orderType": "market",
"quantity": round(qty, 6), "leverage": 1
})
else:
# 负费率套利：做空现货（需要在现货市场借币，或用现货杠杆）
# 简化处理：这里假设支持现货做空，实际需要确认交易所是否支持
spot_order = intoquant_request("POST", "/order/create", {
"exchange": EXCHANGE, "symbol": SYMBOL, "type": "margin",
"side": "sell", "orderType": "market", "quantity": round(qty, 6)
})
# 做多永续
futures_order = intoquant_request("POST", "/order/create", {
"exchange": EXCHANGE, "symbol": SYMBOL, "type": "futures",
"futuresType": "perpetual", "side": "buy", "orderType": "market",
"quantity": round(qty, 6), "leverage": 1
})
if spot_order and futures_order and spot_order.get("code") == 0 and futures_order.get("code") == 0:
log("INFO", f"开仓成功！数量: {qty:.6f} BTC")
state.position = direction
state.spot_qty = qty
state.futures_qty = qty
state.entry_funding = funding_rate
state.entry_spot_price = spot_price
return True
return False
def close_position():
"""平仓"""
log("INFO", f"触发平仓信号，当前持仓: {state.position}，开始平仓...")
spot_price, _ = get_current_prices()
if state.position == 1:
# 平正费率套利：卖出现货 + 平永续空头
spot_order = intoquant_request("POST", "/order/create", {
"exchange": EXCHANGE, "symbol": SYMBOL, "type": "spot",
"side": "sell", "orderType": "market", "quantity": round(state.spot_qty, 6)
})
futures_order = intoquant_request("POST", "/order/create", {
"exchange": EXCHANGE, "symbol": SYMBOL, "type": "futures",
"futuresType": "perpetual", "side": "buy", "orderType": "market",
"quantity": round(state.futures_qty, 6), "reduceOnly": True
})
else:
# 平负费率套利：买入还券 + 平永续多头
spot_order = intoquant_request("POST", "/order/create", {
"exchange": EXCHANGE, "symbol": SYMBOL, "type": "margin",
"side": "buy", "orderType": "market", "quantity": round(state.spot_qty, 6)
})
futures_order = intoquant_request("POST", "/order/create", {
"exchange": EXCHANGE, "symbol": SYMBOL, "type": "futures",
"futuresType": "perpetual", "side": "sell", "orderType": "market",
"quantity": round(state.futures_qty, 6), "reduceOnly": True
})
if spot_order and futures_order and spot_order.get("code") == 0 and futures_order.get("code") == 0:
log("INFO", "平仓成功！")
state.position = 0
state.spot_qty = 0
state.futures_qty = 0
return True
return False
def risk_check(account_value):
"""风控检查"""
if account_value > state.peak_value:
state.peak_value = account_value
dd_exceeded, drawdown = check_drawdown(account_value, state.peak_value, MAX_DRAWDOWN_LIMIT)
if dd_exceeded:
log("WARNING", f"最大回撤超限: {drawdown:.2%}，强制平仓")
if state.position != 0:
close_position()
return False
today = time.strftime("%Y-%m-%d")
if state.last_check_date != today:
state.today_start_value = account_value
state.last_check_date = today
loss_exceeded, daily_loss = check_daily_loss(state.today_start_value, account_value, DAILY_LOSS_LIMIT)
if loss_exceeded:
log("WARNING", f"单日亏损超限: {daily_loss:.2%}，今日暂停交易")
if state.position != 0:
close_position()
return False
# 价格波动止损：检查对冲头寸的价格偏离
if state.position != 0:
spot_price, futures_price = get_current_prices()
if spot_price and state.entry_spot_price > 0:
price_change = abs(spot_price - state.entry_spot_price) / state.entry_spot_price
if price_change > PRICE_STOP_LOSS:
log("WARNING", f"价格波动止损！现货价格变动: {price_change:.2%}，超过阈值: {PRICE_STOP_LOSS:.2%}")
close_position()
return False
return True
# ==================== 主循环 ====================
def main():
log("INFO", "=" * 50)
log("INFO", "资金费率套利策略启动")
log("INFO", f"正费率开仓阈值: {FUNDING_LONG_THRESHOLD:.4%}")
log("INFO", f"负费率开仓阈值: {FUNDING_SHORT_THRESHOLD:.4%}")
log("INFO", "=" * 50)
while True:
try:
account_value = get_account_value()
if account_value is None:
time.sleep(CHECK_INTERVAL)
continue
if not risk_check(account_value):
time.sleep(CHECK_INTERVAL)
continue
# 获取当前资金费率
funding_rate = get_funding_rate(SYMBOL, EXCHANGE)
if funding_rate is None:
time.sleep(CHECK_INTERVAL)
continue
log("INFO", f"当前资金费率: {funding_rate:.4%}，持仓: {state.position}")
if state.position == 0:
# 空仓：检查开仓信号
if funding_rate > FUNDING_LONG_THRESHOLD:
open_position(funding_rate, 1, account_value)
elif funding_rate < FUNDING_SHORT_THRESHOLD:
open_position(funding_rate, -1, account_value)
else:
# 持仓：检查平仓信号
if abs(funding_rate) < abs(FUNDING_CLOSE_THRESHOLD):
close_position()
time.sleep(CHECK_INTERVAL)
except Exception as e:
log("ERROR", f"主循环异常: {e}")
time.sleep(CHECK_INTERVAL)
if __name__ == "__main__":
main()
资金费率套利策略使用说明：
资金费率的预测：资金费率每8小时结算一次，但费率是提前计算的（通常在结算前1小时左右确定）。你可以在a-sig.com上获取"预测资金费率"（predicted funding rate），在结算前就布局，而不是等结算后才反应。
持仓时间：资金费率套利不需要长期持仓。通常在费率高的时候开仓，等费率回归中性后平仓，持仓时间从几小时到几天不等。不要为了等下一次结算而长期持有，因为价格波动风险会累积。
价格波动风险：虽然是对冲头寸，但现货和永续的价格不会完全同步，特别是在剧烈波动时，可能出现"基差扩大"导致对冲头寸浮亏。示例中设置了5%的价格波动止损，实盘中可以根据币种的波动率调整（BTC可以设3%-5%，山寨币需要设更高或避免交易）。
负费率套利的实现：负费率套利需要做空现货，这需要交易所支持现货杠杆（margin trading）或借币。不是所有交易所都支持，也不是所有币种都能借到。在开仓前要确认可借币数量和借币利率。如果借币利率太高（如年化>10%），可能会侵蚀资金费收益。
多币种布局：单一币种的资金费率机会有限，可以同时监控BTC、ETH、SOL等多个主流币种，哪个币种费率高就做哪个。a-sig.com提供了全市场资金费率排行榜，可以快速发现高费率机会。
跨所套利利用同一币种在不同交易所的价格差异，在低价交易所买入，同时在高价交易所卖出，等价差收敛后平仓。
"""
文件：strategy_cross_exchange_arbitrage.py
功能：跨交易所套利策略（简化版）
策略逻辑：
1. 实时监控同一币种在两个交易所的价差
2. 当价差率 > 开仓阈值（覆盖手续费+滑点+缓冲）时，在低价所买入，高价所卖出
3. 当价差率 < 平仓阈值时，平仓获利
4. 管理两个交易所的资金，确保两边都有足够的保证金
"""
import time
from yixin_common import (
log, get_spread, intoquant_request,
check_drawdown, check_daily_loss
)
# ==================== 策略参数 ====================
SYMBOL = "BTC/USDT"
EXCHANGE_A = "binance" # 交易所A
EXCHANGE_B = "okx" # 交易所B
SPREAD_OPEN_THRESHOLD = 0.004 # 价差率开仓阈值：0.4%（需要覆盖两边手续费约0.2%+滑点0.1%+缓冲0.1%）
SPREAD_CLOSE_THRESHOLD = 0.001 # 价差率平仓阈值：0.1%
POSITION_SIZE_RATIO = 0.6 # 仓位比例（跨所需要两边都有资金，所以比例低一些）
MAX_DRAWDOWN_LIMIT = 0.06 # 最大回撤：6%
DAILY_LOSS_LIMIT = 0.03 # 单日亏损：3%
CHECK_INTERVAL = 10 # 检查间隔：10秒（跨所价差变化快，需要高频监控）
# ==================== 策略状态 ====================
class StrategyState:
def __init__(self):
self.position = 0 # 0=空仓, 1=A买B卖, -1=B买A卖
self.qty_a = 0 # 交易所A的持仓数量
self.qty_b = 0 # 交易所B的持仓数量
self.entry_spread = 0 # 开仓时的价差率
self.peak_value = 0
self.today_start_value = 0
self.last_check_date = None
state = StrategyState()
# ==================== 核心函数 ====================
def get_total_account_value():
"""获取两个交易所的总资产合计"""
total = 0
for ex in [EXCHANGE_A, EXCHANGE_B]:
result = intoquant_request("GET", "/account/total-value", {"exchange": ex})
if result and result.get("code") == 0:
total += float(result["data"]["totalValue"])
return total if total > 0 else None
def get_exchange_price(exchange):
"""获取指定交易所的最新价格"""
result = intoquant_request("GET", "/market/ticker",
{"symbol": SYMBOL, "exchange": exchange, "type": "spot"})
if result and result.get("code") == 0:
return float(result["data"]["last"])
return None
def check_balance(exchange, side, qty, price):
"""
检查交易所是否有足够的余额
side=buy: 检查USDT余额是否足够买入qty
side=sell: 检查币的余额是否足够卖出qty
"""
result = intoquant_request("GET", "/account/balances", {"exchange": exchange})
if result and result.get("code") == 0:
balances = result["data"]
if side == "buy":
usdt_balance = float(balances.get("USDT", {}).get("available", 0))
return usdt_balance >= qty * price * 1.01 # 留1%缓冲
else:
coin = SYMBOL.split("/")[0]
coin_balance = float(balances.get(coin, {}).get("available", 0))
return coin_balance >= qty * 1.01
return False
def open_position(spread_rate, direction, account_value):
"""
开仓
direction=1: A价格低，B价格高 -> A买入，B卖出
direction=-1: B价格低，A价格高 -> B买入，A卖出
"""
log("INFO", f"触发开仓信号，价差率: {spread_rate:.4%}，方向: {'A买B卖' if direction==1 else 'B买A卖'}")
price_a = get_exchange_price(EXCHANGE_A)
price_b = get_exchange_price(EXCHANGE_B)
if price_a is None or price_b is None:
return False
# 计算交易数量（用总资金的一部分，两边各一半）
position_value = account_value * POSITION_SIZE_RATIO
buy_price = price_a if direction == 1 else price_b
qty = (position_value / 2) / buy_price
# 确定买卖方向
if direction == 1:
buy_ex, sell_ex = EXCHANGE_A, EXCHANGE_B
else:
buy_ex, sell_ex = EXCHANGE_B, EXCHANGE_A
# 检查余额
if not check_balance(buy_ex, "buy", qty, buy_price):
log("WARNING", f"{buy_ex} USDT余额不足，无法开仓")
return False
if not check_balance(sell_ex, "sell", qty, buy_price):
log("WARNING", f"{sell_ex} 币余额不足，无法开仓")
return False
# 买入
buy_order = intoquant_request("POST", "/order/create", {
"exchange": buy_ex, "symbol": SYMBOL, "type": "spot",
"side": "buy", "orderType": "market", "quantity": round(qty, 6)
})
# 卖出
sell_order = intoquant_request("POST", "/order/create", {
"exchange": sell_ex, "symbol": SYMBOL, "type": "spot",
"side": "sell", "orderType": "market", "quantity": round(qty, 6)
})
if buy_order and sell_order and buy_order.get("code") == 0 and sell_order.get("code") == 0:
log("INFO", f"开仓成功！买入: {buy_ex} {qty:.6f}，卖出: {sell_ex} {qty:.6f}")
state.position = direction
state.qty_a = qty if direction == 1 else -qty # 正数表示多头，负数表示空头
state.qty_b = -qty if direction == 1 else qty
state.entry_spread = spread_rate
return True
return False
def close_position():
"""平仓：反向操作"""
log("INFO", f"触发平仓信号，当前持仓: {state.position}，开始平仓...")
if state.position == 1:
# A买B卖 -> 平仓：A卖出，B买入
sell_ex, buy_ex = EXCHANGE_A, EXCHANGE_B
sell_qty = state.qty_a
buy_qty = abs(state.qty_b)
else:
# B买A卖 -> 平仓：B卖出，A买入
sell_ex, buy_ex = EXCHANGE_B, EXCHANGE_A
sell_qty = state.qty_b
buy_qty = abs(state.qty_a)
# 卖出
sell_order = intoquant_request("POST", "/order/create", {
"exchange": sell_ex, "symbol": SYMBOL, "type": "spot",
"side": "sell", "orderType": "market", "quantity": round(sell_qty, 6)
})
# 买入
buy_order = intoquant_request("POST", "/order/create", {
"exchange": buy_ex, "symbol": SYMBOL, "type": "spot",
"side": "buy", "orderType": "market", "quantity": round(buy_qty, 6)
})
if sell_order and buy_order and sell_order.get("code") == 0 and buy_order.get("code") == 0:
log("INFO", "平仓成功！")
state.position = 0
state.qty_a = 0
state.qty_b = 0
return True
return False
def rebalance_funds():
"""
资金再平衡：当两个交易所的资金不平衡时，划转资金
这是跨所套利的重要环节，确保两边都有足够的资金
简化版：实际中需要考虑转账时间、手续费、链上拥堵等
"""
# 获取两个交易所的USDT余额
balances = {}
for ex in [EXCHANGE_A, EXCHANGE_B]:
result = intoquant_request("GET", "/account/balances", {"exchange": ex})
if result and result.get("code") == 0:
balances[ex] = float(result["data"].get("USDT", {}).get("available", 0))
if len(balances) == 2:
total = sum(balances.values())
target = total / 2
# 如果某个交易所余额低于目标的70%，触发再平衡
for ex in [EXCHANGE_A, EXCHANGE_B]:
if balances[ex] < target * 0.7:
other = EXCHANGE_B if ex == EXCHANGE_A else EXCHANGE_A
transfer_amount = (balances[other] - target) * 0.8 # 转80%的超额部分
if transfer_amount > 100: # 至少转100U
log("INFO", f"触发资金再平衡：从{other}划转{transfer_amount:.2f} USDT到{ex}")
# 调用转账接口（实际中需要确认链上地址、网络等）
# intoquant_request("POST", "/account/transfer", {...})
# 注意：跨所转账需要时间，期间可能无法交易，需要暂停策略
def risk_check(account_value):
if account_value > state.peak_value:
state.peak_value = account_value
dd_exceeded, drawdown = check_drawdown(account_value, state.peak_value, MAX_DRAWDOWN_LIMIT)
if dd_exceeded:
log("WARNING", f"最大回撤超限: {drawdown:.2%}，强制平仓")
if state.position != 0:
close_position()
return False
today = time.strftime("%Y-%m-%d")
if state.last_check_date != today:
state.today_start_value = account_value
state.last_check_date = today
loss_exceeded, daily_loss = check_daily_loss(state.today_start_value, account_value, DAILY_LOSS_LIMIT)
if loss_exceeded:
log("WARNING", f"单日亏损超限: {daily_loss:.2%}，今日暂停交易")
if state.position != 0:
close_position()
return False
return True
# ==================== 主循环 ====================
def main():
log("INFO", "=" * 50)
log("INFO", f"跨所套利策略启动: {EXCHANGE_A} <-> {EXCHANGE_B}")
log("INFO", f"开仓价差阈值: {SPREAD_OPEN_THRESHOLD:.2%}")
log("INFO", "=" * 50)
while True:
try:
account_value = get_total_account_value()
if account_value is None:
time.sleep(CHECK_INTERVAL)
continue
if not risk_check(account_value):
time.sleep(CHECK_INTERVAL)
continue
# 资金再平衡检查（每小时检查一次，简化为每360次循环）
# 实际中可以用时间判断
# 获取当前价差
spread, spread_rate = get_spread(SYMBOL, EXCHANGE_A, EXCHANGE_B)
if spread_rate is None:
time.sleep(CHECK_INTERVAL)
continue
log("INFO", f"当前价差率: {spread_rate:.4%}，持仓: {state.position}")
if state.position == 0:
# 空仓：检查开仓信号
if spread_rate > SPREAD_OPEN_THRESHOLD:
# A价格低，B价格高
open_position(spread_rate, 1, account_value)
elif spread_rate < -SPREAD_OPEN_THRESHOLD:
# B价格低，A价格高
open_position(spread_rate, -1, account_value)
else:
# 持仓：检查平仓信号
if state.position == 1 and spread_rate < SPREAD_CLOSE_THRESHOLD:
close_position()
elif state.position == -1 and spread_rate > -SPREAD_CLOSE_THRESHOLD:
close_position()
time.sleep(CHECK_INTERVAL)
except Exception as e:
log("ERROR", f"主循环异常: {e}")
time.sleep(CHECK_INTERVAL)
if __name__ == "__main__":
main()
跨所套利策略使用说明：
价差阈值的设定：开仓阈值必须覆盖所有成本：
资金管理是核心：跨所套利需要在两个交易所都有资金。如果一个交易所的资金用完了，就无法开仓。需要：
执行速度很重要：跨所价差往往在几秒到几十秒内就会消失。需要：
单边风险：如果一边下单成功另一边失败（如网络问题、余额不足），就会产生单边敞口，面临价格波动风险。需要：
交易所选择：不是所有交易所组合都适合跨所套利。需要考虑：
配对统计套利利用两个高度相关、存在协整关系的币种之间的价差均值回归特性，当价差偏离均值时做收敛交易。
"""
文件：strategy_pairs_trading.py
功能：配对统计套利策略（简化版）
策略逻辑：
1. 选择一对存在协整关系的币种（如ETH和BTC）
2. 实时计算配对价差和Z-Score
3. 当Z-Score > 开仓阈值（如2.0）时，做空价差（做空A+做多B）
4. 当Z-Score < -开仓阈值时，做多价差（做多A+做空B）
5. 当Z-Score回归0附近时平仓
6. 协整关系检验和止损
"""
import time
import numpy as np
import pandas as pd
from yixin_common import (
log, get_klines, intoquant_request, calculate_zscore,
check_drawdown, check_daily_loss
)
from statsmodels.tsa.stattools import coint # 需要安装：pip install statsmodels
# ==================== 策略参数 ====================
SYMBOL_A = "ETH/USDT" # 配对币种A
SYMBOL_B = "BTC/USDT" # 配对币种B
EXCHANGE = "binance"
INTERVAL = "1h" # K线周期：1小时
LOOKBACK_WINDOW = 60 # 滚动窗口：60根K线（60小时）
Z_OPEN_THRESHOLD = 2.0 # Z-Score开仓阈值
Z_CLOSE_THRESHOLD = 0.2 # Z-Score平仓阈值（接近0时平仓）
Z_STOP_LOSS = 3.5 # Z-Score止损阈值（超过3.5说明可能协整断裂）
POSITION_SIZE_RATIO = 0.7
MAX_DRAWDOWN_LIMIT = 0.10
DAILY_LOSS_LIMIT = 0.04
COINTEST_PVALUE = 0.05 # 协整检验p值阈值
CHECK_INTERVAL = 300 # 检查间隔：5分钟（跟K线周期匹配）
# ==================== 策略状态 ====================
class StrategyState:
def __init__(self):
self.position = 0 # 0=空仓, 1=做多价差(多A空B), -1=做空价差(空A多B)
self.qty_a = 0
self.qty_b = 0
self.entry_zscore = 0
self.hedge_ratio = 1.0 # 对冲比例β
self.peak_value = 0
self.today_start_value = 0
self.last_check_date = None
self.last_calc_time = None # 上次计算Z-Score的时间
state = StrategyState()
# ==================== 核心函数 ====================
def get_account_value():
result = intoquant_request("GET", "/account/total-value")
if result and result.get("code") == 0:
return float(result["data"]["totalValue"])
return None
def calculate_hedge_ratio_and_spread(price_a, price_b):
"""
计算对冲比例和配对价差
用线性回归：price_a = α + β * price_b + ε
β就是对冲比例，残差ε就是价差
"""
# 确保两个序列长度相同
min_len = min(len(price_a), len(price_b))
price_a = price_a[-min_len:]
price_b = price_b[-min_len:]
# 线性回归计算对冲比例
# 简化版：用numpy的polyfit
coeffs = np.polyfit(price_b, price_a, 1)
beta = coeffs[0] # 斜率就是对冲比例
# 计算价差
spread = price_a - beta * price_b
return beta, spread
def check_cointegration(price_a, price_b):
"""
检验两个价格序列是否存在协整关系
返回：(是否协整, p值)
"""
min_len = min(len(price_a), len(price_b))
price_a = price_a[-min_len:]
price_b = price_b[-min_len:]
try:
score, pvalue, _ = coint(price_a, price_b)
return pvalue < COINTEST_PVALUE, pvalue
except Exception as e:
log("ERROR", f"协整检验失败: {e}")
return False, 1.0
def get_current_zscore():
"""
获取当前的Z-Score，同时更新对冲比例
"""
# 获取两个币种的K线数据
df_a = get_klines(SYMBOL_A, INTERVAL, LOOKBACK_WINDOW + 10)
df_b = get_klines(SYMBOL_B, INTERVAL, LOOKBACK_WINDOW + 10)
if df_a is None or df_b is None:
return None, None, None
# 对齐时间索引
df_a, df_b = df_a.align(df_b, join="inner", axis=0)
if len(df_a) < LOOKBACK_WINDOW:
log("WARNING", f"K线数据不足，当前{len(df_a)}根，需要{LOOKBACK_WINDOW}根")
return None, None, None
close_a = df_a["close"].values
close_b = df_b["close"].values
# 协整检验
is_cointegrated, pvalue = check_cointegration(close_a, close_b)
if not is_cointegrated:
log("WARNING", f"配对不存在协整关系，p值: {pvalue:.4f}，暂停交易")
return None, None, None
# 计算对冲比例和价差
beta, spread = calculate_hedge_ratio_and_spread(close_a, close_b)
# 计算Z-Score
spread_series = pd.Series(spread)
zscore_series = calculate_zscore(spread_series, LOOKBACK_WINDOW)
current_zscore = zscore_series.iloc[-1]
log("INFO", f"对冲比例β: {beta:.4f}，当前Z-Score: {current_zscore:.4f}，协整p值: {pvalue:.4f}")
return current_zscore, beta, (close_a[-1], close_b[-1])
def open_position(zscore, beta, prices, account_value):
"""
开仓
zscore>0: 价差偏高 -> 做空价差（做空A + 做多B）
zscore<0: 价差偏低 -> 做多价差（做多A + 做空B）
"""
direction = -1 if zscore > 0 else 1
log("INFO", f"触发开仓信号，Z-Score: {zscore:.4f}，方向: {'做空价差' if direction==-1 else '做多价差'}")
price_a, price_b = prices
# 计算仓位（用总资金的一部分）
position_value = account_value * POSITION_SIZE_RATIO
# 按对冲比例分配资金
# A的价值 = beta * B的价值（确保组合是市场中性的）
value_b = position_value / (1 + beta)
value_a = position_value - value_b
qty_a = value_a / price_a
qty_b = value_b / price_b
if direction == 1:
# 做多价差：做多A + 做空B
order_a = intoquant_request("POST", "/order/create", {
"exchange": EXCHANGE, "symbol": SYMBOL_A, "type": "spot",
"side": "buy", "orderType": "market", "quantity": round(qty_a, 6)
})
order_b = intoquant_request("POST", "/order/create", {
"exchange": EXCHANGE, "symbol": SYMBOL_B, "type": "futures",
"futuresType": "perpetual", "side": "sell", "orderType": "market",
"quantity": round(qty_b, 6), "leverage": 1
})
else:
# 做空价差：做空A（需要现货杠杆或借币）+ 做多B
order_a = intoquant_request("POST", "/order/create", {
"exchange": EXCHANGE, "symbol": SYMBOL_A, "type": "margin",
"side": "sell", "orderType": "market", "quantity": round(qty_a, 6)
})
order_b = intoquant_request("POST", "/order/create", {
"exchange": EXCHANGE, "symbol": SYMBOL_B, "type": "futures",
"futuresType": "perpetual", "side": "buy", "orderType": "market",
"quantity": round(qty_b, 6), "leverage": 1
})
if order_a and order_b and order_a.get("code") == 0 and order_b.get("code") == 0:
log("INFO", f"开仓成功！A: {qty_a:.6f}, B: {qty_b:.6f}")
state.position = direction
state.qty_a = qty_a if direction == 1 else -qty_a
state.qty_b = -qty_b if direction == 1 else qty_b
state.entry_zscore = zscore
state.hedge_ratio = beta
return True
return False
def close_position():
"""平仓"""
log("INFO", f"触发平仓信号，当前持仓: {state.position}，开始平仓...")
if state.position == 1:
# 平多价差：卖出A + 平B空头
order_a = intoquant_request("POST", "/order/create", {
"exchange": EXCHANGE, "symbol": SYMBOL_A, "type": "spot",
"side": "sell", "orderType": "market", "quantity": round(abs(state.qty_a), 6)
})
order_b = intoquant_request("POST", "/order/create", {
"exchange": EXCHANGE, "symbol": SYMBOL_B, "type": "futures",
"futuresType": "perpetual", "side": "buy", "orderType": "market",
"quantity": round(abs(state.qty_b), 6), "reduceOnly": True
})
else:
# 平空价差：买入还A + 平B多头
order_a = intoquant_request("POST", "/order/create", {
"exchange": EXCHANGE, "symbol": SYMBOL_A, "type": "margin",
"side": "buy", "orderType": "market", "quantity": round(abs(state.qty_a), 6)
})
order_b = intoquant_request("POST", "/order/create", {
"exchange": EXCHANGE, "symbol": SYMBOL_B, "type": "futures",
"futuresType": "perpetual", "side": "sell", "orderType": "market",
"quantity": round(abs(state.qty_b), 6), "reduceOnly": True
})
if order_a and order_b and order_a.get("code") == 0 and order_b.get("code") == 0:
log("INFO", "平仓成功！")
state.position = 0
state.qty_a = 0
state.qty_b = 0
return True
return False
def risk_check(account_value, current_zscore):
"""风控检查"""
if account_value > state.peak_value:
state.peak_value = account_value
dd_exceeded, drawdown = check_drawdown(account_value, state.peak_value, MAX_DRAWDOWN_LIMIT)
if dd_exceeded:
log("WARNING", f"最大回撤超限: {drawdown:.2%}，强制平仓")
if state.position != 0:
close_position()
return False
today = time.strftime("%Y-%m-%d")
if state.last_check_date != today:
state.today_start_value = account_value
state.last_check_date = today
loss_exceeded, daily_loss = check_daily_loss(state.today_start_value, account_value, DAILY_LOSS_LIMIT)
if loss_exceeded:
log("WARNING", f"单日亏损超限: {daily_loss:.2%}，今日暂停交易")
if state.position != 0:
close_position()
return False
# Z-Score止损：如果Z-Score超过止损阈值，说明可能协整断裂，强制平仓
if state.position != 0 and current_zscore is not None:
if abs(current_zscore) > Z_STOP_LOSS:
log("WARNING", f"Z-Score止损！当前Z-Score: {current_zscore:.4f}，超过阈值: {Z_STOP_LOSS}，可能协整断裂")
close_position()
return False
return True
# ==================== 主循环 ====================
def main():
log("INFO", "=" * 50)
log("INFO", f"配对统计套利策略启动: {SYMBOL_A} - {SYMBOL_B}")
log("INFO", f"Z-Score开仓阈值: {Z_OPEN_THRESHOLD}")
log("INFO", f"Z-Score平仓阈值: {Z_CLOSE_THRESHOLD}")
log("INFO", "=" * 50)
while True:
try:
account_value = get_account_value()
if account_value is None:
time.sleep(CHECK_INTERVAL)
continue
# 获取当前Z-Score（每根K线计算一次，简化为每次循环都算）
current_zscore, beta, prices = get_current_zscore()
if current_zscore is None:
time.sleep(CHECK_INTERVAL)
continue
if not risk_check(account_value, current_zscore):
time.sleep(CHECK_INTERVAL)
continue
if state.position == 0:
# 空仓：检查开仓信号
if current_zscore > Z_OPEN_THRESHOLD:
open_position(current_zscore, beta, prices, account_value)
elif current_zscore < -Z_OPEN_THRESHOLD:
open_position(current_zscore, beta, prices, account_value)
else:
# 持仓：检查平仓信号
if abs(current_zscore) < Z_CLOSE_THRESHOLD:
close_position()
time.sleep(CHECK_INTERVAL)
except Exception as e:
log("ERROR", f"主循环异常: {e}")
time.sleep(CHECK_INTERVAL)
if __name__ == "__main__":
main()
配对统计套利策略使用说明：
配对选择是关键：不是任意两个币种都能做配对套利。需要满足：
对冲比例的动态调整：示例中每次都重新计算对冲比例β，这是正确的做法。因为两个币种的价格关系会随时间变化，固定的对冲比例会导致对冲不精确，产生额外的风险。更高级的方法是用卡尔曼滤波动态估计时变的对冲比例，效果更好。
协整关系的监控：配对套利最大的风险是协整断裂——两个币种的长期均衡关系被打破（如其中一个币种出现黑天鹅事件），价差不回归反而持续扩大。需要：
做空的实现：配对套利通常需要做空其中一个币种。可以用：
参数优化：
网格套利在设定的价格区间内，设置一系列价格档位，价格每下跌一个档位买入，每上涨一个档位卖出，通过区间内的高抛低吸获利。
"""
文件：strategy_grid_trading.py
功能：网格套利策略（简化版，现货网格）
策略逻辑：
1. 设定价格区间（上界、下界）和网格数量
2. 在区间内等间距（或等比例）设置网格线
3. 价格每下跌一格买入固定数量，每上涨一格卖出固定数量
4. 价格突破区间上界或下界时，暂停网格或调整区间
5. 严格的资金管理和风控
"""
import time
import numpy as np
from yixin_common import (
log, get_klines, intoquant_request, calculate_atr,
check_drawdown, check_daily_loss
)
# ==================== 策略参数 ====================
SYMBOL = "BTC/USDT"
EXCHANGE = "binance"
# 网格参数（也可以用ATR自动计算，见下方函数）
GRID_UPPER = 32000.0 # 网格上界
GRID_LOWER = 28000.0 # 网格下界
GRID_COUNT = 20 # 网格数量（20格 = 21条网格线）
GRID_TYPE = "arithmetic" # 网格类型：arithmetic(等差), geometric(等比)
POSITION_SIZE_RATIO = 0.8 # 总仓位比例
MAX_DRAWDOWN_LIMIT = 0.15 # 最大回撤：15%（网格策略回撤可能较大）
DAILY_LOSS_LIMIT = 0.05 # 单日亏损：5%
CHECK_INTERVAL = 30 # 检查间隔：30秒
# ==================== 策略状态 ====================
class StrategyState:
def __init__(self):
self.grid_lines = [] # 网格线价格列表
self.grid_qty = 0 # 每格交易数量
self.holdings = {} # 持仓记录：{网格线索引: 买入数量}
self.total_cost = 0 # 总买入成本
self.total_holdings = 0 # 总持仓数量
self.peak_value = 0
self.today_start_value = 0
self.last_check_date = None
self.initialized = False # 是否已初始化网格
state = StrategyState()
# ==================== 核心函数 ====================
def get_account_value():
result = intoquant_request("GET", "/account/total-value")
if result and result.get("code") == 0:
return float(result["data"]["totalValue"])
return None
def get_current_price():
result = intoquant_request("GET", "/market/ticker",
{"symbol": SYMBOL, "exchange": EXCHANGE, "type": "spot"})
if result and result.get("code") == 0:
return float(result["data"]["last"])
return None
def calculate_grid_lines():
"""
计算网格线价格
等差网格：每格价格差相等
等比网格：每格价格比例相等（适合价格波动大的标的）
"""
lines = []
if GRID_TYPE == "arithmetic":
step = (GRID_UPPER - GRID_LOWER) / GRID_COUNT
for i in range(GRID_COUNT + 1):
lines.append(GRID_LOWER + i * step)
else: # geometric
ratio = (GRID_UPPER / GRID_LOWER) ** (1 / GRID_COUNT)
for i in range(GRID_COUNT + 1):
lines.append(GRID_LOWER * (ratio ** i))
return lines
def calculate_grid_qty(account_value):
"""
计算每格的交易数量
假设所有网格都被触发买入，总资金不超过仓位限制
"""
total_budget = account_value * POSITION_SIZE_RATIO
# 平均价格
avg_price = (GRID_UPPER + GRID_LOWER) / 2
# 总买入数量 = 总预算 / 平均价格
total_qty = total_budget / avg_price
# 每格数量 = 总数量 / 网格数
grid_qty = total_qty / GRID_COUNT
return grid_qty
def initialize_grid(account_value):
"""初始化网格：计算网格线、每格数量、初始底仓"""
log("INFO", "初始化网格...")
state.grid_lines = calculate_grid_lines()
state.grid_qty = calculate_grid_qty(account_value)
current_price = get_current_price()
if current_price is None:
return False
# 初始底仓：如果当前价格在区间内，买入当前价格以下所有网格的底仓
# 简化处理：先买入50%的底仓
initial_qty = state.grid_qty * GRID_COUNT * 0.5
initial_cost = initial_qty * current_price
if initial_cost > account_value * POSITION_SIZE_RATIO:
initial_qty = (account_value * POSITION_SIZE_RATIO) / current_price
log("INFO", f"买入初始底仓: {initial_qty:.6f} {SYMBOL.split('/')[0]}，价格: {current_price:.2f}")
order = intoquant_request("POST", "/order/create", {
"exchange": EXCHANGE, "symbol": SYMBOL, "type": "spot",
"side": "buy", "orderType": "market", "quantity": round(initial_qty, 6)
})
if order and order.get("code") == 0:
state.total_holdings = initial_qty
state.total_cost = initial_qty * current_price
state.initialized = True
log("INFO", f"网格初始化完成！网格数: {GRID_COUNT}，每格数量: {state.grid_qty:.6f}")
log("INFO", f"网格区间: [{GRID_LOWER:.2f}, {GRID_UPPER:.2f}]")
return True
else:
log("ERROR", "初始底仓买入失败")
return False
def find_grid_index(price):
"""找到价格所在的网格线索引"""
for i in range(len(state.grid_lines) - 1):
if state.grid_lines[i] <= price < state.grid_lines[i + 1]:
return i
if price >= state.grid_lines[-1]:
return len(state.grid_lines) - 1
return 0
def check_grid_trades(prev_price, current_price):
"""
检查从上一次检查到现在，价格穿越了哪些网格线，执行相应的买卖
"""
if prev_price == current_price:
return
# 确定价格移动方向
if current_price > prev_price:
# 价格上涨：检查穿越了哪些网格线，执行卖出
for i in range(len(state.grid_lines)):
line_price = state.grid_lines[i]
if prev_price < line_price <= current_price:
# 向上穿越了这条网格线，执行卖出
if i in state.holdings and state.holdings[i] > 0:
sell_qty = state.holdings[i]
log("INFO", f"网格卖出！价格穿越网格线{i}: {line_price:.2f}，卖出: {sell_qty:.6f}")
order = intoquant_request("POST", "/order/create", {
"exchange": EXCHANGE, "symbol": SYMBOL, "type": "spot",
"side": "sell", "orderType": "market", "quantity": round(sell_qty, 6)
})
if order and order.get("code") == 0:
state.holdings[i] = 0
state.total_holdings -= sell_qty
# 计算利润：卖出价 - 买入价
# 简化处理，实际需要记录每笔买入成本
else:
# 价格下跌：检查穿越了哪些网格线，执行买入
for i in range(len(state.grid_lines) - 1, -1, -1):
line_price = state.grid_lines[i]
if prev_price > line_price >= current_price:
# 向下穿越了这条网格线，执行买入
if i not in state.holdings or state.holdings[i] == 0:
buy_qty = state.grid_qty
log("INFO", f"网格买入！价格穿越网格线{i}: {line_price:.2f}，买入: {buy_qty:.6f}")
order = intoquant_request("POST", "/order/create", {
"exchange": EXCHANGE, "symbol": SYMBOL, "type": "spot",
"side": "buy", "orderType": "market", "quantity": round(buy_qty, 6)
})
if order and order.get("code") == 0:
state.holdings[i] = buy_qty
state.total_holdings += buy_qty
state.total_cost += buy_qty * line_price
def check_range_breakout(current_price):
"""
检查价格是否突破网格区间
突破上界：全部卖出，暂停网格
突破下界：暂停买入，等待价格回归或调整区间
"""
if current_price > GRID_UPPER:
log("WARNING", f"价格突破网格上界！当前: {current_price:.2f}，上界: {GRID_UPPER:.2f}")
# 全部卖出
if state.total_holdings > 0:
log("INFO", f"全部卖出持仓: {state.total_holdings:.6f}")
order = intoquant_request("POST", "/order/create", {
"exchange": EXCHANGE, "symbol": SYMBOL, "type": "spot",
"side": "sell", "orderType": "market", "quantity": round(state.total_holdings, 6)
})
if order and order.get("code") == 0:
state.holdings = {}
state.total_holdings = 0
# 暂停策略，等待人工调整区间
log("WARNING", "网格暂停，请调整区间后重新启动")
return True # 返回True表示需要暂停
elif current_price < GRID_LOWER:
log("WARNING", f"价格突破网格下界！当前: {current_price:.2f}，下界: {GRID_LOWER:.2f}")
# 下界突破不强制卖出（可能继续持有等反弹），但暂停新的买入
log("WARNING", "价格低于网格下界，暂停买入，持有现有仓位等待反弹")
# 实际中可以设置一个止损线，跌破下界一定比例后止损
return False # 继续运行但不买入
return False
def risk_check(account_value):
if account_value > state.peak_value:
state.peak_value = account_value
dd_exceeded, drawdown = check_drawdown(account_value, state.peak_value, MAX_DRAWDOWN_LIMIT)
if dd_exceeded:
log("WARNING", f"最大回撤超限: {drawdown:.2%}，全部卖出止损")
if state.total_holdings > 0:
order = intoquant_request("POST", "/order/create", {
"exchange": EXCHANGE, "symbol": SYMBOL, "type": "spot",
"side": "sell", "orderType": "market", "quantity": round(state.total_holdings, 6)
})
if order and order.get("code") == 0:
state.holdings = {}
state.total_holdings = 0
return False
today = time.strftime("%Y-%m-%d")
if state.last_check_date != today:
state.today_start_value = account_value
state.last_check_date = today
loss_exceeded, daily_loss = check_daily_loss(state.today_start_value, account_value, DAILY_LOSS_LIMIT)
if loss_exceeded:
log("WARNING", f"单日亏损超限: {daily_loss:.2%}，今日暂停交易")
return False
return True
# ==================== 主循环 ====================
def main():
log("INFO", "=" * 50)
log("INFO", "网格套利策略启动")
log("INFO", f"网格区间: [{GRID_LOWER:.2f}, {GRID_UPPER:.2f}]")
log("INFO", f"网格数量: {GRID_COUNT}，类型: {GRID_TYPE}")
log("INFO", "=" * 50)
prev_price = None
while True:
try:
account_value = get_account_value()
if account_value is None:
time.sleep(CHECK_INTERVAL)
continue
# 初始化网格
if not state.initialized:
if not initialize_grid(account_value):
time.sleep(CHECK_INTERVAL)
continue
if not risk_check(account_value):
time.sleep(CHECK_INTERVAL)
continue
current_price = get_current_price()
if current_price is None:
time.sleep(CHECK_INTERVAL)
continue
log("INFO", f"当前价格: {current_price:.2f}，持仓: {state.total_holdings:.6f}")
# 检查区间突破
should_pause = check_range_breakout(current_price)
if should_pause:
break # 暂停策略
# 检查网格交易
if prev_price is not None:
check_grid_trades(prev_price, current_price)
prev_price = current_price
time.sleep(CHECK_INTERVAL)
except Exception as e:
log("ERROR", f"主循环异常: {e}")
time.sleep(CHECK_INTERVAL)
if __name__ == "__main__":
main()
网格套利策略使用说明：
网格区间的设定：这是网格策略最关键的参数。区间太窄，价格容易突破，频繁暂停；区间太宽，网格密度低，交易机会少，资金利用率低。建议：
网格数量的设定：网格数量决定了每格的间距和每格的资金量。
等差vs等比网格：
初始底仓：示例中买入了50%的底仓。实际中可以根据当前价格位置调整：
区间突破的处理：
合约网格vs现货网格：示例是现货网格，你也可以用合约网格：
最后，提供一些代码部署和运维的建议，帮助你把这些策略从"能运行"变成"稳定盈利"。
# 1. 准备云服务器（推荐配置：2核4G以上，Linux系统）
# 推荐选择交易所服务器所在地区的节点（如AWS东京、阿里云新加坡），降低延迟
# 2. 安装Python环境
sudo apt update
sudo apt install python3 python3-pip python3-venv
# 3. 创建项目目录和虚拟环境
mkdir -p /opt/yixin_strategy
cd /opt/yixin_strategy
python3 -m venv venv
source venv/bin/activate
# 4. 安装依赖
pip install requests pandas numpy ccxt python-dotenv statsmodels
# 5. 上传代码和.env文件
# .env文件内容示例：
# A_SIG_API_KEY=your_a_sig_api_key
# INTOQUANT_API_KEY=your_intoquant_api_key
# INTOQUANT_API_SECRET=your_intoquant_api_secret
# 注意：.env文件包含敏感信息，不要上传到Git，设置文件权限600
# 6. 用systemd管理进程（开机自启、崩溃自动重启）
# 创建 /etc/systemd/system/grid-strategy.service
# [Unit]
# Description=Yixin Grid Trading Strategy
# After=network.target
#
# [Service]
# Type=simple
# User=root
# WorkingDirectory=/opt/yixin_strategy
# ExecStart=/opt/yixin_strategy/venv/bin/python /opt/yixin_strategy/strategy_grid_trading.py
# Restart=always
# RestartSec=10
#
# [Install]
# WantedBy=multi-user.target
# 启用和启动
sudo systemctl daemon-reload
sudo systemctl enable grid-strategy
sudo systemctl start grid-strategy
# 查看日志
sudo journalctl -u grid-strategy -f
# 建议使用Python的logging模块，而不是简单的print
# 日志同时输出到控制台和文件，文件按天轮转，保留30天
import logging
from logging.handlers import TimedRotatingFileHandler
def setup_logger(name, log_file, level=logging.INFO):
formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
handler = TimedRotatingFileHandler(log_file, when="midnight", interval=1, backupCount=30)
handler.setFormatter(formatter)
console_handler = logging.StreamHandler()
console_handler.setFormatter(formatter)
logger = logging.getLogger(name)
logger.setLevel(level)
logger.addHandler(handler)
logger.addHandler(console_handler)
return logger
# 使用
logger = setup_logger("grid_strategy", "/var/log/yixin/grid.log")
logger.info("策略启动")
logger.warning("风险告警")
logger.error("交易失败")
# 关键事件（开仓、平仓、风控触发、系统异常）需要实时通知
# 可以用Telegram Bot、企业微信、钉钉、邮件等方式
import requests
def send_telegram_alert(bot_token, chat_id, message):
"""发送Telegram告警"""
url = f"https://api.telegram.org/bot{bot_token}/sendMessage"
params = {
"chat_id": chat_id,
"text": message,
"parse_mode": "Markdown"
}
try:
requests.post(url, params=params, timeout=10)
except Exception as e:
print(f"发送告警失败: {e}")
# 使用示例
# send_telegram_alert(BOT_TOKEN, CHAT_ID, "⚠️ *风控告警*\n最大回撤超限: 8.5%，已强制平仓")
# 1. 策略代码用Git管理，每次修改都提交，方便回滚
git init
git add .
git commit -m "初始版本"
# 修改后
git commit -am "优化网格参数"
# 回滚到上一个版本
git revert HEAD
# 2. 配置文件和参数单独管理，不要硬编码在代码里
# 用config.yaml或.env管理配置，修改配置不需要改代码
# 3. 定期备份交易记录和账户数据
# 可以每天导出交易历史，保存到数据库或文件
# 用于后续的绩效分析和策略优化
在把任何策略投入实盘前，请确认以下事项：
本附录提供了五大核心套利策略的简化版Python代码，涵盖期现套利、资金费率套利、跨所套利、配对统计套利、网格套利。每个策略都包含完整的策略逻辑、信号计算、下单执行、风控检查和主循环，并有详细的中文注释。
这些代码是"简化版"，重点是展示策略逻辑和API调用方式。实盘使用时，你还需要：
壹信量化平台提供了完整的策略开发和部署工具链：
如果你在策略开发或部署过程中遇到问题，可以访问intoquant.com获取专业的技术支持和定制服务。
最后再次提醒：量化交易有风险，实盘前请充分回测和模拟盘验证，控制仓位，做好风控。本代码仅供学习参考，不构成投资建议。
祝你的量化套利之路顺利！
⚠️ 风险提示：本书内容仅为量化研究与知识分享，不构成任何投资建议。套利交易存在基差、费率、流动性、平台与极端行情等风险，历史表现不代表未来收益。投资有风险，入市需谨慎，请自主决策、量力而行。
💡 键盘 ←/→ 翻章 · T 键切换目录