Python装饰器在量化交易中的实战:让代码从＂能用＂到＂优雅＂

import time
from functools import wraps

def risk_check(func):
   """下单前风控检查装饰器"""
   @wraps(func)
   def wrapper(self, *args, **kwargs):
       # 检查1：交易时段
       if not self.is_trading_hour():
           raise RuntimeError(f"非交易时段，当前时间：{time.strftime('%H:%M:%S')}")
       
       # 检查2：保证金充足
       available = self.get_available_margin()
       required = kwargs.get('required_margin', 0)
       if available < required:
           raise RuntimeError(f"保证金不足，可用：{available}，需要：{required}")
       
       # 检查3：当日亏损上限
       today_pnl = self.get_today_pnl()
       if today_pnl < -self.max_daily_loss:
           raise RuntimeError(f"已达当日最大亏损{self.max_daily_loss}，当前亏损：{today_pnl}")
       
       return func(self, *args, **kwargs)
   return wrapper
 

class MomentumStrategy:
   def __init__(self):
       self.max_daily_loss = 50000  # 当日最大亏损
   
   @risk_check
   def place_order(self, symbol, volume, price):
       # 真正的下单逻辑在这里
       print(f"下单成功：{symbol} {volume}手 @{price}")
       return {"status": "success", "order_id": self.generate_order_id()}
 

import time
import logging
from functools import wraps

logger = logging.getLogger('quant.performance')

def timing(func):
   """性能计时装饰器，输出函数名和执行时长"""
   @wraps(func)
   def wrapper(*args, **kwargs):
       start = time.perf_counter()
       result = func(*args, **kwargs)
       elapsed = time.perf_counter() - start
       
       # 超过1秒的函数记录warning级别
       level = logging.WARNING if elapsed > 1.0 else logging.DEBUG
       logger.log(level, f"[{func.__name__}] 执行耗时: {elapsed:.3f}秒")
       
       return result
   return wrapper
 

from abc import ABC, abstractmethod

class Strategy(ABC):
   """策略基类（抽象接口）"""
   
   @abstractmethod
   def generate_signals(self, price_data: dict) -> list:
       """生成交易信号，子类必须实现"""
       pass
   
   @abstractmethod
   def get_position_size(self, signal: dict, price: float) -> int:
       """计算仓位大小，子类必须实现"""
       pass

class TrendStrategy(Strategy):
   """趋势跟踪策略"""
   def __init__(self, lookback=20, threshold=0.02):
       self.lookback = lookback
       self.threshold = threshold
   
   def generate_signals(self, price_data):
       # 具体实现省略，返回信号列表
       pass
   
   def get_position_size(self, signal, price):
       return signal['strength'] * 10000 // price

class MeanRevStrategy(Strategy):
   """均值回归策略"""
   def __init__(self, window=30, std_threshold=2.0):
       self.window = window
       self.std_threshold = std_threshold
   
   def generate_signals(self, price_data):
       # 具体实现省略，返回信号列表
       pass
   
   def get_position_size(self, signal, price):
       return abs(signal['zscore']) * 5000 // price
 

class StrategyRunner:
   """策略调度器：只依赖Strategy接口，不关心具体策略"""
   
   def __init__(self, strategy: Strategy):
       self.strategy = strategy
   
   def run(self, price_data):
       signals = self.strategy.generate_signals(price_data)
       for sig in signals:
           if sig['action'] == 'BUY':
               size = self.strategy.get_position_size(sig, sig['price'])
               self.executor.execute_buy(sig['symbol'], size, sig['price'])
           elif sig['action'] == 'SELL':
               size = self.strategy.get_position_size(sig, sig['price'])
               self.executor.execute_sell(sig['symbol'], size, sig['price'])
   
   def switch_strategy(self, new_strategy: Strategy):
       """运行时切换策略，一行代码搞定"""
       self.strategy = new_strategy
       logger.info(f"策略切换成功：{new_strategy.__class__.__name__}")
 

@log_execution
@timing
@risk_check
@retry(max_attempts=3, delay=1.0)
def fetch_market_data(symbol: str, start_date: str, end_date: str) -> pd.DataFrame:
   """获取行情数据：自动日志+计时+风控+重试"""
   return akshare_stock_zh_a_hist(symbol=symbol, start_date=start_date, end_date=end_date)