场景：从一次惨痛的API限额错误说起
2024年3月，我负责的一个量化交易团队遭遇了致命打击。凌晨2点17分，系统在执行止损订单时突然中断，错误日志显示：429 Too Many Requests - Rate limit exceeded。更糟糕的是，由于缺乏熔断机制，程序在重试循环中持续触发API限额，最终导致当夜亏损超过47%。这个案例揭示了量化交易合规开发中最容易被忽视的环节——不仅是策略本身的合规，更是基础设施、API调用、风险管理流程的系统性合规。
本文将深入探讨加密货币量化交易的核心合规要点，涵盖数据使用规范、回测标准、以及风控框架的完整搭建方案。这些经验来自我参与多个大型量化平台架构设计的实战总结。
加密货币量化交易合规框架概览
量化交易的合规性不是事后检查，而是贯穿整个系统生命周期的系统性工程。一个完整的合规框架需要涵盖以下几个核心维度：
- 数据合规：数据来源的合法性、数据使用的授权范围、数据存储的安全标准
- 回测合规：历史数据的质量保证、回测环境的真实性、回测结果的统计显著性
- 风控合规：仓位管理规则、止损熔断机制、异常交易监控
- API调用合规：速率限制遵守、请求签名规范、错误处理机制
- 审计追溯：完整的交易日志、决策树记录、风险事件追踪
第一部分：数据使用规范与合规策略
数据源合规性检查清单
在中国市场进行加密货币量化交易，数据合规是首要考量。根据监管要求，量化团队必须确保所有市场数据来源合法、使用授权明确。以下是数据合规的核心检查项：
- 数据提供商是否持有相关金融服务牌照
- 数据使用范围是否明确约定（研究/回测/实盘）
- 历史数据的完整性和准确性是否经过第三方验证
- 高频数据的时序精度是否满足策略需求（毫秒级/微秒级）
- 数据存储是否符合个人信息保护法规要求
多交易所数据聚合的合规架构
实际项目中，我们通常需要聚合多个交易所的数据进行套利策略开发。以下是一个合规的数据聚合架构示例：
"""
加密货币量化交易数据聚合模块 - 合规版
支持多交易所数据源整合，内置速率限制和错误处理
"""
import asyncio
import aiohttp
import logging
from datetime import datetime, timedelta
from typing import Dict, List, Optional
from dataclasses import dataclass, field
from enum import Enum
import hashlib
import json
class DataSource(Enum):
BINANCE = "binance"
OKX = "okx"
BYBIT = "bybit"
HOLYSHEEP_API = "holysheep" # 统一API网关
class RateLimitStatus(Enum):
GREEN = "green" # 可用
YELLOW = "yellow" # 接近限额
RED = "red" # 触发限制
@dataclass
class APIKeyConfig:
"""API密钥配置 - 符合合规存储要求"""
exchange: str
api_key: str
api_secret: str
passphrase: Optional[str] = None # OKX等需要
permissions: List[str] = field(default_factory=list) # 权限范围
@dataclass
class RateLimitState:
"""速率限制状态追踪"""
requests_made: int = 0
requests_limit: int = 1200
reset_timestamp: datetime = field(default_factory=datetime.now)
status: RateLimitStatus = RateLimitStatus.GREEN
class ComplianceDataAggregator:
"""
合规数据聚合器
特点：
- 内置速率限制器，避免触发API限额
- 完整的请求日志，满足审计要求
- 多数据源冗余，自动故障转移
"""
def __init__(
self,
api_keys: List[APIKeyConfig],
holysheep_base_url: str = "https://api.holysheep.ai/v1",
holysheep_api_key: str = "YOUR_HOLYSHEEP_API_KEY"
):
self.api_keys = {k.exchange: k for k in api_keys}
self.rate_limiters: Dict[str, RateLimitState] = {}
self.logger = logging.getLogger(__name__)
# HolySheep统一API网关 - 可替代高昂的交易所官方API费用
# ¥1=$1，节省85%+成本，延迟<50ms
self.holysheep_base_url = holysheep_base_url
self.holysheep_api_key = holysheep_api_key
# 合规审计日志
self.audit_log: List[Dict] = []
self._init_rate_limiters()
def _init_rate_limiters(self):
"""初始化各交易所速率限制器"""
default_limits = {
"binance": RateLimitState(requests_limit=1200), # 1分钟
"okx": RateLimitState(requests_limit=600), # 1分钟
"bybit": RateLimitState(requests_limit=600), # 1分钟
}
self.rate_limiters.update(default_limits)
def _log_audit(self, action: str, details: Dict):
"""记录合规审计日志"""
entry = {
"timestamp": datetime.now().isoformat(),
"action": action,
"details": details,
"hash": hashlib.sha256(
json.dumps(details, sort_keys=True).encode()
).hexdigest()[:16]
}
self.audit_log.append(entry)
# 保留最近10000条审计记录
if len(self.audit_log) > 10000:
self.audit_log = self.audit_log[-10000:]
def _check_rate_limit(self, exchange: str) -> RateLimitStatus:
"""检查速率限制状态"""
state = self.rate_limiters.get(exchange)
if not state:
return RateLimitStatus.GREEN
now = datetime.now()
if now >= state.reset_timestamp:
# 重置计数器
state.requests_made = 0
state.reset_timestamp = now + timedelta(minutes=1)
usage_ratio = state.requests_made / state.requests_limit
if usage_ratio >= 0.9:
state.status = RateLimitStatus.RED
elif usage_ratio >= 0.7:
state.status = RateLimitStatus.YELLOW
else:
state.status = RateLimitStatus.GREEN
return state.status
async def _request_with_retry(
self,
session: aiohttp.ClientSession,
exchange: str,
method: str,
url: str,
headers: Dict = None,
params: Dict = None,
max_retries: int = 3
) -> Dict:
"""
带重试机制的请求方法
内置速率限制检查和错误处理
"""
for attempt in range(max_retries):
# 检查速率限制
status = self._check_rate_limit(exchange)
if status == RateLimitStatus.RED:
self.logger.warning(f"Rate limit RED for {exchange}, backing off...")
await asyncio.sleep(5 * (attempt + 1)) # 指数退避
try:
async with session.request(
method, url, headers=headers, params=params, timeout=aiohttp.ClientTimeout(total=10)
) as response:
# 更新速率限制状态
if exchange in self.rate_limiters:
self.rate_limiters[exchange].requests_made += 1
self._log_audit("api_request", {
"exchange": exchange,
"method": method,
"url": url,
"status": response.status,
"attempt": attempt + 1
})
if response.status == 200:
return await response.json()
elif response.status == 429:
# 速率限制触发 - 记录并等待
self.logger.error(f"429 Rate Limited: {exchange}")
self.rate_limiters[exchange].status = RateLimitStatus.RED
await asyncio.sleep(60) # 等待限速窗口重置
continue
elif response.status == 401:
self.logger.critical(f"401 Unauthorized: {exchange}")
raise PermissionError(f"API认证失败: {exchange}")
else:
raise aiohttp.ClientResponseError(
request_info=response.request_info,
history=[],
status=response.status
)
except aiohttp.ClientError as e:
self.logger.error(f"Request failed (attempt {attempt + 1}): {e}")
if attempt == max_retries - 1:
raise
await asyncio.sleep(2 ** attempt)
raise RuntimeError(f"Max retries exceeded for {exchange}")
async def fetch_kline_data(
self,
exchange: str,
symbol: str,
interval: str = "1h",
limit: int = 1000,
start_time: Optional[int] = None,
end_time: Optional[int] = None
) -> List[Dict]:
"""
获取K线数据 - 合规实现
自动处理速率限制和错误恢复
"""
if exchange == "binance":
url = f"https://api.binance.com/api/v3/klines"
params = {
"symbol": symbol,
"interval": interval,
"limit": limit
}
elif exchange == "okx":
url = f"https://www.okx.com/api/v5/market/candles"
params = {
"instId": symbol,
"bar": interval,
"limit": str(limit)
}
else:
raise ValueError(f"Unsupported exchange: {exchange}")
async with aiohttp.ClientSession() as session:
data = await self._request_with_retry(
session, exchange, "GET", url, params=params
)
return self._normalize_kline_data(exchange, data)
def _normalize_kline_data(self, exchange: str, raw_data: List) -> List[Dict]:
"""标准化不同交易所的数据格式"""
normalized = []
for candle in raw_data:
if exchange == "binance":
normalized.append({
"timestamp": candle[0],
"open": float(candle[1]),
"high": float(candle[2]),
"low": float(candle[3]),
"close": float(candle[4]),
"volume": float(candle[5]),
"close_time": candle[6]
})
elif exchange == "okx":
normalized.append({
"timestamp": int(float(candle[0]) * 1000),
"open": float(candle[1]),
"high": float(candle[2]),
"low": float(candle[3]),
"close": float(candle[4]),
"volume": float(candle[5]),
"close_time": int(float(candle[6]) * 1000)
})
return normalized
def get_audit_log(self, limit: int = 100) -> List[Dict]:
"""获取审计日志 - 用于合规检查"""
return self.audit_log[-limit:]
使用示例
async def main():
# 合规配置 - API密钥应从安全的密钥管理服务获取
api_keys = [
APIKeyConfig(
exchange="binance",
api_key="YOUR_BINANCE_API_KEY",
api_secret="YOUR_BINANCE_SECRET",
permissions=["market_data", "spot_trading"]
),
APIKeyConfig(
exchange="okx",
api_key="YOUR_OKX_API_KEY",
api_secret="YOUR_OKX_SECRET",
passphrase="YOUR_PASSPHRASE",
permissions=["market_data"]
)
]
aggregator = ComplianceDataAggregator(
api_keys=api_keys,
holysheep_base_url="https://api.holysheep.ai/v1",
holysheep_api_key="YOUR_HOLYSHEEP_API_KEY"
)
# 获取数据
klines = await aggregator.fetch_kline_data(
exchange="binance",
symbol="BTCUSDT",
interval="1h",
limit=500
)
print(f"获取到 {len(klines)} 条K线数据")
print(f"合规审计日志: {len(aggregator.get_audit_log())} 条记录")
if __name__ == "__main__":
asyncio.run(main())这段代码的核心设计理念是：将合规性内嵌到每一个API调用中，而非事后检查。速率限制追踪、审计日志、错误重试机制都是量化交易系统稳定运行的基础保障。
第二部分：回测规范与合规标准
回测合规的核心挑战
回测是量化策略开发的核心环节，也是最容易出现"虚假繁荣"的环节。常见的回测偏差包括：
- 前视偏差：使用未来才能获得的信息进行当前决策
- 幸存者偏差：仅选择历史中存在至今的标的
- 流动性偏差：假设大额订单可以瞬间以市价成交
- 参数过拟合：过度优化策略参数导致实盘失效
- 市场冲击：未考虑大单交易对价格的影响
合规回测框架实现
"""
加密货币量化交易回测引擎 - 合规版
内置防止常见回测偏差的机制
"""
import numpy as np
import pandas as pd
from datetime import datetime, timedelta
from typing import Dict, List, Tuple, Optional, Callable
from dataclasses import dataclass, field
from enum import Enum
import logging
from abc import ABC, abstractmethod
class BacktestBias(Enum):
"""回测偏差类型"""
LOOK_AHEAD = "look_ahead"
SURVIVORSHIP = "survivorship"
LIQUIDITY = "liquidity"
OVERFITTING = "overfitting"
MARKET_IMPACT = "market_impact"
@dataclass
class BacktestConfig:
"""回测配置"""
initial_capital: float = 100000.0
commission_rate: float = 0.001 # 0.1% 手续费
slippage_model: str = "fixed" # 滑点模型: fixed, percentage, volume_based
slippage_rate: float = 0.0005 # 固定滑点 0.05%
min_order_size: float = 10.0 # 最小订单金额
max_position_pct: float = 0.2 # 最大仓位比例 20%
confidence_level: float = 0.95 # 置信区间
benchmark: str = "BTCUSDT"
@dataclass
class BacktestResult:
"""回测结果"""
total_return: float
sharpe_ratio: float
max_drawdown: float
win_rate: float
profit_factor: float
calmar_ratio: float
trade_count: int
equity_curve: pd.Series
daily_returns: pd.Series
bias_check: Dict[str, bool] = field(default_factory=dict)
statistics: Dict = field(default_factory=dict)
@dataclass
class TradeRecord:
"""交易记录"""
timestamp: int
symbol: str
side: str # BUY or SELL
price: float
quantity: float
commission: float
slippage: float
realized_pnl: float
cumulative_pnl: float
class BiasDetector:
"""
回测偏差检测器
自动检测常见的回测偏差
"""
def __init__(self):
self.logger = logging.getLogger(__name__)
def detect_look_ahead_bias(
self,
df: pd.DataFrame,
signal_col: str,
price_col: str = "close"
) -> Tuple[bool, str]:
"""
检测前视偏差
通过检查信号和价格的时间戳关系
"""
if signal_col not in df.columns or price_col not in df.columns:
return False, "Missing columns"
# 检查是否有信号在价格更新前产生
# 正确做法：信号在t时刻产生，对应t+1的价格
if "signal_time" in df.columns and "price_time" in df.columns:
time_diff = df["signal_time"] - df["price_time"]
if (time_diff > 0).any():
return True, f"Detected {time_diff[time_diff > 0].sum()} look-ahead violations"
return False, "No look-ahead bias detected"
def detect_liquidity_bias(
self,
df: pd.DataFrame,
order_size_col: str,
volume_col: str,
threshold: float = 0.1
) -> Tuple[bool, str]:
"""
检测流动性偏差
订单量不应超过市场成交量的固定比例
"""
if order_size_col not in df.columns or volume_col not in df.columns:
return False, "Missing columns"
participation_rate = df[order_size_col] / df[volume_col]
violations = (participation_rate > threshold).sum()
if violations > 0:
return True, f"{violations} orders exceed {threshold*100}% participation rate"
return False, "No liquidity bias detected"
def check_all_biases(self, df: pd.DataFrame, config: BacktestConfig) -> Dict[str, bool]:
"""综合偏差检查"""
results = {}
# 前视偏差检查
has_look_ahead, msg = self.detect_look_ahead_bias(df, "signal", "close")
results["look_ahead_bias"] = has_look_ahead
if has_look_ahead:
self.logger.warning(f"Look-ahead bias: {msg}")
# 流动性偏差检查
if "order_size" in df.columns and "volume" in df.columns:
has_liquidity, msg = self.detect_liquidity_bias(df, "order_size", "volume")
results["liquidity_bias"] = has_liquidity
if has_liquidity:
self.logger.warning(f"Liquidity bias: {msg}")
return results
class ComplianceBacktestEngine:
"""
合规回测引擎
特点：
- 内置偏差检测机制
- 真实的交易成本模拟
- 滑点模型
- 分阶段验证
"""
def __init__(
self,
config: BacktestConfig,
holysheep_base_url: str = "https://api.holysheep.ai/v1",
holysheep_api_key: str = "YOUR_HOLYSHEEP_API_KEY"
):
self.config = config
self.logger = logging.getLogger(__name__)
self.bias_detector = BiasDetector()
# HolySheep API用于数据验证和信号生成
self.holysheep_base_url = holysheep_base_url
self.holysheep_api_key = holysheep_api_key
self.trades: List[TradeRecord] = []
self.equity_curve: List[float] = []
self.positions: Dict[str, float] = {}
self.cash: float = config.initial_capital
def _apply_slippage(
self,
price: float,
quantity: float,
side: str,
current_volume: float = float('inf')
) -> Tuple[float, float]:
"""
应用滑点模型
支持多种滑点计算方式
"""
if self.config.slippage_model == "fixed":
# 固定滑点
slippage_rate = self.config.slippage_rate
elif self.config.slippage_model == "percentage":
# 百分比滑点（基于订单量相对于市场深度）
slippage_rate = min(quantity / max(current_volume, 1) * 0.01, 0.01)
else:
slippage_rate = self.config.slippage_rate
if side == "BUY":
execution_price = price * (1 + slippage_rate)
else:
execution_price = price * (1 - slippage_rate)
slippage_cost = abs(execution_price - price) * quantity
return execution_price, slippage_cost
def _calculate_commission(self, price: float, quantity: float) -> float:
"""计算手续费"""
return price * quantity * self.config.commission_rate
def _validate_order(
self,
symbol: str,
side: str,
quantity: float,
price: float
) -> Tuple[bool, str]:
"""
订单验证 - 防止合规问题
"""
order_value = price * quantity
# 最小订单金额检查
if order_value < self.config.min_order_size:
return False, f"Order value {order_value} below minimum {self.config.min_order_size}"
# 最大仓位检查
if side == "BUY":
new_position_value = self.positions.get(symbol, 0) * price + order_value
total_equity = self.cash + sum(
self.positions.get(s, 0) * price for s in self.positions
)
if total_equity > 0 and new_position_value / total_equity > self.config.max_position_pct:
return False, f"Position would exceed max {self.config.max_position_pct*100}%"
# 余额检查
if side == "BUY" and order_value > self.cash:
return False, f"Insufficient cash: need {order_value}, have {self.cash}"
return True, "Valid"
def execute_trade(
self,
timestamp: int,
symbol: str,
side: str,
quantity: float,
price: float,
volume: float = float('inf')
):
"""执行交易"""
# 订单验证
is_valid, msg = self._validate_order(symbol, side, quantity, price)
if not is_valid:
self.logger.warning(f"Order rejected: {msg}")
return None
# 应用滑点
execution_price, slippage = self._apply_slippage(price, quantity, side, volume)
# 计算手续费
commission = self._calculate_commission(execution_price, quantity)
# 执行交易
if side == "BUY":
cost = execution_price * quantity + commission + slippage
self.cash -= cost
self.positions[symbol] = self.positions.get(symbol, 0) + quantity
else:
revenue = execution_price * quantity - commission - slippage
self.cash += revenue
self.positions[symbol] = self.positions.get(symbol, 0) - quantity
# 记录交易
trade = TradeRecord(
timestamp=timestamp,
symbol=symbol,
side=side,
price=execution_price,
quantity=quantity,
commission=commission,
slippage=slippage,
realized_pnl=0, # 在平仓时计算
cumulative_pnl=self.cash + sum(
self.positions.get(s, 0) * price for s in self.positions
) - self.config.initial_capital
)
self.trades.append(trade)
self.logger.debug(
f"Trade executed: {side} {quantity} {symbol} @ {execution_price}, "
f"commission: {commission}, slippage: {slippage}"
)
return trade
def calculate_metrics(self) -> BacktestResult:
"""计算回测指标"""
equity_series = pd.Series(self.equity_curve)
daily_returns = equity_series.pct_change().dropna()
# 基本指标
total_return = (equity_series.iloc[-1] / equity_series.iloc[0] - 1) * 100
# 夏普比率
if daily_returns.std() > 0:
sharpe_ratio = np.sqrt(252) * daily_returns.mean() / daily_returns.std()
else:
sharpe_ratio = 0
# 最大回撤
cummax = equity_series.cummax()
drawdown = (equity_series - cummax) / cummax
max_drawdown = abs(drawdown.min()) * 100
# 卡玛比率
calmar_ratio = total_return / max_drawdown if max_drawdown > 0 else 0
# 交易统计
trades_df = pd.DataFrame([vars(t) for t in self.trades])
win_rate = (trades_df[trades_df['side'] == 'SELL']['realized_pnl'] > 0).mean() if len(trades_df) > 0 else 0
return BacktestResult(
total_return=total_return,
sharpe_ratio=sharpe_ratio,
max_drawdown=max_drawdown,
win_rate=win_rate * 100,
profit_factor=0, # 需要平仓数据计算
calmar_ratio=calmar_ratio,
trade_count=len(self.trades),
equity_curve=equity_series,
daily_returns=daily_returns,
bias_check={},
statistics={
"avg_trade_value": (trades_df['price'] * trades_df['quantity']).mean() if len(trades_df) > 0 else 0,
"total_commission": trades_df['commission'].sum() if len(trades_df) > 0 else 0,
"total_slippage": trades_df['slippage'].sum() if len(trades_df) > 0 else 0,
}
)
def run(
self,
data: pd.DataFrame,
strategy_func: Callable
) -> BacktestResult:
"""
运行回测
参数:
data: 包含OHLCV数据的DataFrame
strategy_func: 策略函数，输入历史数据，输出信号
"""
self.logger.info(f"Starting backtest with {len(data)} bars")
# 初始化
self.trades = []
self.equity_curve = [self.config.initial_capital]
self.positions = {}
self.cash = self.config.initial_capital
look_back = 20 # 最小回看窗口
for i in range(look_back, len(data)):
# 获取历史窗口
window = data.iloc[:i]
current = data.iloc[i]
# 生成信号
signal = strategy_func(window)
# 执行交易
if signal and signal.get('action') in ['BUY', 'SELL']:
symbol = signal.get('symbol', 'BTCUSDT')
quantity = signal.get('quantity', 0)
if quantity > 0:
self.execute_trade(
timestamp=current.name if isinstance(current.name, int) else i,
symbol=symbol,
side=signal['action'],
quantity=quantity,
price=current['close'],
volume=current.get('volume', float('inf'))
)
# 更新权益
portfolio_value = self.cash + sum(
self.positions.get(s, 0) * current['close'] for s in self.positions
)
self.equity_curve.append(portfolio_value)
# 计算最终指标
result = self.calculate_metrics()
# 偏差检查
result.bias_check = self.bias_detector.check_all_biases(data, self.config)
self.logger.info(
f"Backtest completed: Return={result.total_return:.2f}%, "
f"Sharpe={result.sharpe_ratio:.2f}, MaxDD={result.max_drawdown:.2f}%"
)
return result
使用示例
def example_strategy(window: pd.DataFrame) -> dict:
"""
示例策略：简单移动平均交叉
仅使用历史数据，不存在前视偏差
"""
if len(window) < 20:
return None
ma5 = window['close'].rolling(5).mean().iloc[-1]
ma20 = window['close'].rolling(20).mean().iloc[-1]
ma5_prev = window['close'].rolling(5).mean().iloc[-2]
ma20_prev = window['close'].rolling(20).mean().iloc[-2]
# 金叉买入
if ma5_prev <= ma20_prev and ma5 > ma20:
return {
'action': 'BUY',
'symbol': 'BTCUSDT',
'quantity': 0.1 # 固定手数
}
# 死叉卖出
elif ma5_prev >= ma20_prev and ma5 < ma20:
return {
'action': 'SELL',
'symbol': 'BTCUSDT',
'quantity': 0.1
}
return None
运行回测
if __name__ == "__main__":
logging.basicConfig(level=logging.INFO)
# 回测配置
config = BacktestConfig(
initial_capital=100000.0,
commission_rate=0.001,
slippage_model="fixed",
slippage_rate=0.0005,
min_order_size=10.0,
max_position_pct=0.2
)
engine = ComplianceBacktestEngine(
config=config,
holysheep_base_url="https://api.holysheep.ai/v1",
holysheep_api_key="YOUR_HOLYSHEEP_API_KEY"
)
# 加载数据（需要真实数据）
# data = pd.read_csv('btc_usdt_1h.csv', index_col=0, parse_dates=True)
# result = engine.run(data, example_strategy)
# print(f"Total Return: {result.total_return:.2f}%")
# print(f"Sharpe Ratio: {result.sharpe_ratio:.2f}")
# print(f"Max Drawdown: {result.max_drawdown:.2f}%")
print("Backtest framework initialized successfully")这个回测框架的核心设计原则是：合规不是限制，而是保护。通过内置的偏差检测、交易验证、完整的成本模拟，我们能够在回测阶段就发现潜在问题，避免实盘亏损。
第三部分：风控框架搭建
多层次风控体系设计
一个完善的风控框架需要涵盖以下几个层次：
- 交易前风控：订单验证、仓位检查、资金充足性验证
- 交易中风控：实时监控、异常检测、自动熔断
- 交易后风控：日终对账、绩效归因、风险报告
- 系统性风控：技术故障应对、网络安全、数据备份
完整风控框架实现
"""
加密货币量化交易风控框架 - 生产级实现
多层防护机制，确保交易安全
"""
import asyncio
import logging
from datetime import datetime, time
from typing import Dict, List, Optional, Callable
from dataclasses import dataclass, field
from enum import Enum
from collections import deque
import threading
import time as time_module
class RiskLevel(Enum):
GREEN = "green" # 正常
YELLOW = "yellow" # 警告
ORANGE = "orange" # 风险
RED = "red" # 危险
CRITICAL = "critical" # 紧急熔断
class ViolationType(Enum):
POSITION_LIMIT = "position_limit"
DAILY_LOSS = "daily_loss"
CONCENTRATION = "concentration"
LEVERAGE = "leverage"
WITHDRAWAL = "withdrawal"
RATE_LIMIT = "rate_limit"
API_ERROR = "api_error"
@dataclass
class RiskLimit:
"""风控限额配置"""
max_position_pct: float = 0.25 # 单币种最大仓位25%
max_total_position_pct: float = 0.8 # 总仓位上限80%
max_daily_loss_pct: float = 0.05 # 日内最大亏损5%
max_weekly_loss_pct: float = 0.10 # 周内最大亏损10%
max_concentration_pct: float = 0.40 # 单币种持仓集中度上限40%
max_leverage: float = 3.0 # 最大杠杆倍数
max_orders_per_minute: int = 60 # 每分钟最大订单数
max_order_value: float = 1000000.0 # 单笔最大订单金额
circuit_breaker_threshold: float = 0.08 # 熔断阈值8%
@dataclass
class RiskViolation:
"""风控违规记录"""
timestamp: datetime
violation_type: ViolationType
current_value: float
limit_value: float
severity: RiskLevel
action_taken: str
details: str
@dataclass
class Position:
"""持仓信息"""
symbol: str
quantity: float
avg_entry_price: float
current_price: float
unrealized_pnl: float
timestamp: datetime
class CircuitBreaker:
"""
熔断机制
当亏损达到阈值时自动停止交易
"""
def __init__(self, threshold: float, reset_period: int = 300):
self.threshold = threshold
self.reset_period = reset_period # 秒
self.is_triggered = False
self.triggered_at: Optional[datetime] = None
self.trigger_count = 0
self.total_loss = 0.0
self.initial_capital = 0.0
def update(self, current_capital: float, initial_capital: float):
"""更新熔断状态"""
self.initial_capital = initial_capital
self.total_loss = (initial_capital - current_capital) / initial_capital
if not self.is_triggered:
if self.total_loss >= self.threshold:
self.is_triggered = True
self.triggered_at = datetime.now()
self.trigger_count += 1
return True # 触发熔断
# 检查是否需要重置
if self.is_triggered:
elapsed = (datetime.now() - self.triggered_at).total_seconds()
if elapsed >= self.reset_period:
self.is_triggered = False
self.triggered_at = None
return False
def get_status(self) -> Dict:
return {
"is_triggered": self.is_triggered,
"total_loss_pct": self.total_loss * 100,
"threshold_pct": self.threshold * 100,
"trigger_count": self.trigger_count,
"next_reset": (
(self.triggered