# 高频交易策略算法实例import numpy as np
import pandas as pd
from collections import deque
from typing import Dict, List, Tuple
import time# ==================== 策略1: 做市商策略 (Market Making) ====================
class MarketMakingStrategy:"""做市商策略：通过在买卖两侧同时挂单，赚取买卖价差"""def __init__(self, spread: float = 0.001, order_size: float = 100):self.spread = spread # 价差比例self.order_size = order_size # 每次下单量self.inventory = 0 # 当前持仓self.max_inventory = 1000 # 最大持仓限制def generate_quotes(self, mid_price: float) -> Tuple[float, float]:"""生成买卖报价"""# 根据持仓调整报价，持仓过多时降低买价提高卖价inventory_skew = self.inventory / self.max_inventory * 0.0005bid_price = mid_price * (1 - self.spread / 2 - inventory_skew)ask_price = mid_price * (1 + self.spread / 2 - inventory_skew)return bid_price, ask_pricedef should_quote(self) -> bool:"""判断是否应该报价"""return abs(self.inventory) < self.max_inventorydef update_inventory(self, trade_side: str, quantity: float):"""更新持仓"""if trade_side == 'buy':self.inventory += quantityelif trade_side == 'sell':self.inventory -= quantity# ==================== 策略2: 统计套利策略 (Statistical Arbitrage) ====================
class StatisticalArbitrageStrategy:"""统计套利：基于协整关系的配对交易"""def __init__(self, lookback_period: int = 100, entry_threshold: float = 2.0, exit_threshold: float = 0.5):self.lookback_period = lookback_periodself.entry_threshold = entry_threshold # 入场标准差倍数self.exit_threshold = exit_threshold # 出场标准差倍数self.price_history_a = deque(maxlen=lookback_period)self.price_history_b = deque(maxlen=lookback_period)self.position = 0 # 1: 做多价差, -1: 做空价差, 0: 无持仓def calculate_spread(self, price_a: float, price_b: float, hedge_ratio: float) -> float:"""计算价差"""return price_a - hedge_ratio * price_bdef calculate_zscore(self, current_spread: float, spread_history: List[float]) -> float:"""计算Z-Score"""mean_spread = np.mean(spread_history)std_spread = np.std(spread_history)if std_spread == 0:return 0return (current_spread - mean_spread) / std_spreaddef generate_signal(self, price_a: float, price_b: float, hedge_ratio: float) -> int:"""生成交易信号"""self.price_history_a.append(price_a)self.price_history_b.append(price_b)if len(self.price_history_a) < self.lookback_period:return 0# 计算历史价差spread_history = [self.price_history_a[i] - hedge_ratio * self.price_history_b[i]for i in range(len(self.price_history_a))]current_spread = self.calculate_spread(price_a, price_b, hedge_ratio)zscore = self.calculate_zscore(current_spread, spread_history)# 生成信号if self.position == 0:if zscore > self.entry_threshold:self.position = -1 # 价差过高，做空价差return -1elif zscore < -self.entry_threshold:self.position = 1 # 价差过低，做多价差return 1else:# 平仓逻辑if abs(zscore) < self.exit_threshold:signal = -self.positionself.position = 0return signalreturn 0# ==================== 策略3: 动量策略 (Momentum Strategy) ====================
class MomentumStrategy:"""动量策略：捕捉短期价格趋势"""def __init__(self, fast_period: int = 10, slow_period: int = 30, signal_threshold: float = 0.0005):self.fast_period = fast_periodself.slow_period = slow_periodself.signal_threshold = signal_thresholdself.price_history = deque(maxlen=slow_period)self.position = 0def calculate_ema(self, prices: List[float], period: int) -> float:"""计算指数移动平均"""if len(prices) < period:return np.mean(prices)multiplier = 2 / (period + 1)ema = prices[0]for price in prices[1:]:ema = (price - ema) * multiplier + emareturn emadef generate_signal(self, current_price: float) -> int:"""生成交易信号"""self.price_history.append(current_price)if len(self.price_history) < self.slow_period:return 0prices_list = list(self.price_history)fast_ema = self.calculate_ema(prices_list[-self.fast_period:], self.fast_period)slow_ema = self.calculate_ema(prices_list, self.slow_period)# 计算动量信号momentum = (fast_ema - slow_ema) / slow_emaif momentum > self.signal_threshold and self.position <= 0:self.position = 1return 1 # 买入信号elif momentum < -self.signal_threshold and self.position >= 0:self.position = -1return -1 # 卖出信号return 0# ==================== 策略4: 订单流失衡策略 (Order Flow Imbalance) ====================
class OrderFlowImbalanceStrategy:"""订单流失衡策略：基于买卖订单量的失衡进行交易"""def __init__(self, window_size: int = 50, imbalance_threshold: float = 0.3):self.window_size = window_sizeself.imbalance_threshold = imbalance_thresholdself.buy_volume_history = deque(maxlen=window_size)self.sell_volume_history = deque(maxlen=window_size)def calculate_imbalance(self, buy_volume: float, sell_volume: float) -> float:"""计算订单流失衡度"""total_volume = buy_volume + sell_volumeif total_volume == 0:return 0return (buy_volume - sell_volume) / total_volumedef generate_signal(self, buy_volume: float, sell_volume: float) -> int:"""生成交易信号"""self.buy_volume_history.append(buy_volume)self.sell_volume_history.append(sell_volume)if len(self.buy_volume_history) < self.window_size:return 0# 计算累积订单流失衡total_buy = sum(self.buy_volume_history)total_sell = sum(self.sell_volume_history)imbalance = self.calculate_imbalance(total_buy, total_sell)# 生成信号if imbalance > self.imbalance_threshold:return 1 # 买单占优，买入elif imbalance < -self.imbalance_threshold:return -1 # 卖单占优，卖出return 0# ==================== 策略5: 微观结构策略 (Microstructure Strategy) ====================
class MicrostructureStrategy:"""微观结构策略：基于买卖价差和深度的策略"""def __init__(self, spread_threshold: float = 0.001, depth_ratio_threshold: float = 1.5):self.spread_threshold = spread_thresholdself.depth_ratio_threshold = depth_ratio_thresholddef calculate_spread_ratio(self, bid: float, ask: float) -> float:"""计算价差比例"""mid_price = (bid + ask) / 2return (ask - bid) / mid_pricedef calculate_depth_imbalance(self, bid_depth: float, ask_depth: float) -> float:"""计算深度失衡"""total_depth = bid_depth + ask_depthif total_depth == 0:return 0return (bid_depth - ask_depth) / total_depthdef generate_signal(self, bid: float, ask: float, bid_depth: float, ask_depth: float) -> int:"""生成交易信号"""spread_ratio = self.calculate_spread_ratio(bid, ask)# 价差过大时不交易if spread_ratio > self.spread_threshold:return 0depth_imbalance = self.calculate_depth_imbalance(bid_depth, ask_depth)# 基于深度失衡生成信号if depth_imbalance > 0.3:return 1 # 买单深度大，预期上涨elif depth_imbalance < -0.3:return -1 # 卖单深度大，预期下跌return 0# ==================== 策略管理器 ====================
class StrategyManager:"""策略管理器：统一管理多个策略"""def __init__(self):self.strategies = {}self.signals = {}def add_strategy(self, name: str, strategy):"""添加策略"""self.strategies[name] = strategyself.signals[name] = 0def update_signals(self, market_data: Dict):"""更新所有策略信号"""for name, strategy in self.strategies.items():if isinstance(strategy, MarketMakingStrategy):# 做市商策略特殊处理passelse:# 其他策略更新信号passdef get_combined_signal(self) -> int:"""获取综合信号"""total_signal = sum(self.signals.values())if total_signal > 0:return 1elif total_signal < 0:return -1return 0# ==================== 使用示例 ====================
def example_usage():"""策略使用示例"""# 初始化策略mm_strategy = MarketMakingStrategy(spread=0.001, order_size=100)stat_arb = StatisticalArbitrageStrategy(lookback_period=100)momentum = MomentumStrategy(fast_period=10, slow_period=30)order_flow = OrderFlowImbalanceStrategy(window_size=50)micro = MicrostructureStrategy()# 模拟市场数据mid_price = 100.0# 做市商策略if mm_strategy.should_quote():bid, ask = mm_strategy.generate_quotes(mid_price)print(f"做市商报价 - 买价: {bid:.2f}, 卖价: {ask:.2f}")# 动量策略signal = momentum.generate_signal(mid_price)print(f"动量策略信号: {signal}")# 订单流策略signal = order_flow.generate_signal(buy_volume=1000, sell_volume=800)print(f"订单流策略信号: {signal}")# 微观结构策略signal = micro.generate_signal(bid=99.95, ask=100.05, bid_depth=5000, ask_depth=3000)print(f"微观结构策略信号: {signal}")if __name__ == "__main__":example_usage()
HFT策略系统
│
├── 策略1: MarketMakingStrategy (做市商策略)
│ ├── generate_quotes() - 生成买卖报价
│ ├── should_quote() - 判断是否报价
│ └── update_inventory() - 更新持仓
│
├── 策略2: StatisticalArbitrageStrategy (统计套利)
│ ├── calculate_spread() - 计算价差
│ ├── calculate_zscore() - 计算Z分数
│ └── generate_signal() - 生成交易信号
│
├── 策略3: MomentumStrategy (动量策略)
│ ├── calculate_ema() - 计算指数移动平均
│ └── generate_signal() - 生成交易信号
│
├── 策略4: OrderFlowImbalanceStrategy (订单流失衡)
│ ├── calculate_imbalance() - 计算失衡度
│ └── generate_signal() - 生成交易信号
│
├── 策略5: MicrostructureStrategy (微观结构)
│ ├── calculate_spread_ratio() - 计算价差比例
│ ├── calculate_depth_imbalance() - 计算深度失衡
│ └── generate_signal() - 生成交易信号
│
└── StrategyManager (策略管理器)├── add_strategy() - 添加策略├── update_signals() - 更新信号└── get_combined_signal() - 获取综合信号
