量化策略的代码优化与性能调优：从Python到生产级系统

import cProfile
import pstats
import numpy as np
import pandas as pd
from line_profiler import LineProfiler
from memory_profiler import profile

# 示例：一个典型的因子计算函数
def calculate_momentum_factor(prices, window=20):
    """计算动量因子"""
    returns = prices.pct_change()
    momentum = returns.rolling(window).mean()
    return momentum

# 方法1：cProfile - 函数级性能分析
def profile_with_cprofile():
    prices = pd.DataFrame(np.random.randn(10000, 100))
    
    profiler = cProfile.Profile()
    profiler.enable()
    
    # 待分析代码
    for _ in range(100):
        calculate_momentum_factor(prices)
    
    profiler.disable()
    
    # 输出分析结果
    stats = pstats.Stats(profiler)
    stats.sort_stats('cumtime')  # 按累计时间排序
    stats.print_stats(10)  # 显示前10个最耗时的函数

# 方法2：line_profiler - 行级性能分析
@profile  # line_profiler装饰器
def calculate_factor_with_loop(prices):
    """使用循环计算因子（低效示例）"""
    n = len(prices)
    factor = np.zeros(n)
    
    for i in range(20, n):
        # 糟糕的循环实现
        window_data = prices[i-20:i]
        factor[i] = window_data.mean()
    
    return factor

# 方法3：memory_profiler - 内存使用分析
@profile  # memory_profiler装饰器
def memory_intensive_operation():
    """内存密集型操作"""
    # 读取大文件
    df = pd.read_csv('large_file.csv')
    
    # 创建多个副本（内存泄漏常见原因）
    df_copy1 = df.copy()
    df_copy2 = df.copy()
    
    # 执行计算
    result = df.groupby('symbol').apply(complex_calculation)
    
    return result

import numpy as np
import pandas as pd
import numba
from numba import jit, njit, prange
import time

# 低效示例：Python循环计算移动平均
def moving_average_loop(prices, window):
    """使用循环计算移动平均（慢）"""
    n = len(prices)
    ma = np.zeros(n)
    
    for i in range(window - 1, n):
        ma[i] = np.mean(prices[i - window + 1:i + 1])
    
    return ma

# 优化版本1：pandas内置函数
def moving_average_pandas(prices, window):
    """使用pandas滚动窗口（中）"""
    return pd.Series(prices).rolling(window).mean().values

# 优化版本2：numpy卷积
def moving_average_numpy(prices, window):
    """使用numpy卷积（快）"""
    weights = np.ones(window) / window
    return np.convolve(prices, weights, mode='same')

# 优化版本3：numba JIT编译
@njit(parallel=True, fastmath=True)
def moving_average_numba(prices, window):
    """使用numba JIT编译（最快）"""
    n = len(prices)
    ma = np.zeros(n)
    
    for i in range(window - 1, n):
        total = 0.0
        for j in range(i - window + 1, i + 1):
            total += prices[j]
        ma[i] = total / window
    
    return ma

# 性能对比测试
def benchmark_moving_average():
    """性能基准测试"""
    np.random.seed(42)
    prices = np.random.randn(1000000)  # 100万条数据
    
    # 测试不同实现
    implementations = [
        ("Python循环", moving_average_loop),
        ("pandas滚动", moving_average_pandas),
        ("numpy卷积", moving_average_numpy),
        ("numba JIT", moving_average_numba)
    ]
    
    results = {}
    
    for name, func in implementations:
        start_time = time.time()
        result = func(prices, 20)
        elapsed = time.time() - start_time
        results[name] = elapsed
        print(f"{name}: {elapsed:.4f}秒")
    
    # 输出加速比
    print("\n加速比：")
    baseline = results["Python循环"]
    for name, elapsed in results.items():
        speedup = baseline / elapsed
        print(f"{name}: {speedup:.1f}x")

# 运行基准测试
if __name__ == "__main__":
    benchmark_moving_average()

import numba
from numba import jit, njit, prange
import numpy as np

# 示例1：计算波动率因子（带并行化）
@njit(parallel=True, fastmath=True, cache=True)
def calculate_volatility_parallel(returns, window):
    """
    并行计算滚动波动率
    
    参数:
        returns: 收益率序列
        window: 滚动窗口
    
    返回:
        波动率序列
    """
    n = len(returns)
    volatility = np.zeros(n)
    
    # 并行化外层循环
    for i in prange(window - 1, n):
        # 计算滚动标准差
        sum_sq = 0.0
        for j in range(i - window + 1, i + 1):
            sum_sq += returns[j] ** 2
        
        volatility[i] = np.sqrt(sum_sq / window)
    
    return volatility

# 示例2：布林带策略信号生成（完全JIT编译）
@njit(cache=True)
def bollinger_band_signal(prices, window=20, num_std=2.0):
    """
    计算布林带交易信号
    
    返回:
        1: 买入信号（价格跌破下轨）
        -1: 卖出信号（价格突破上轨）
        0: 无信号
    """
    n = len(prices)
    signals = np.zeros(n)
    
    # 计算均值和标准差
    ma = np.zeros(n)
    std = np.zeros(n)
    
    for i in range(window - 1, n):
        # 计算滚动均值
        total = 0.0
        for j in range(i - window + 1, i + 1):
            total += prices[j]
        ma[i] = total / window
        
        # 计算滚动标准差
        sum_sq = 0.0
        for j in range(i - window + 1, i + 1):
            sum_sq += (prices[j] - ma[i]) ** 2
        std[i] = np.sqrt(sum_sq / window)
    
    # 生成信号
    upper_band = ma + num_std * std
    lower_band = ma - num_std * std
    
    for i in range(window, n):
        if prices[i] < lower_band[i]:
            signals[i] = 1  # 买入
        elif prices[i] > upper_band[i]:
            signals[i] = -1  # 卖出
    
    return signals

# 示例3：避免Numba的坑
@njit
def common_numba_mistakes():
    """Numba常见错误示例"""
    
    # 错误1：使用Python复杂对象
    # my_dict = {}  # ❌ Numba不支持dict
    # my_list = []  # ❌ 动态list不支持
    
    # 正确：使用numpy array
    my_array = np.zeros(100)  # ✅
    
    # 错误2：调用不支持的pandas函数
    # df = pd.DataFrame()  # ❌ Numba中不能用pandas
    
    # 正确：只使用numpy和纯数值计算
    for i in range(100):
        my_array[i] = i ** 2
    
    return my_array

import multiprocessing as mp
from concurrent.futures import ProcessPoolExecutor, ThreadPoolExecutor
import pandas as pd
import numpy as np
from functools import partial

# 场景：参数优化（网格搜索）
def optimize_parameters_parallel():
    """并行参数优化"""
    
    def backtest_strategy(data, params):
        """单次回测（耗时操作）"""
        window, threshold = params
        
        # 模拟耗时计算
        signals = generate_signals(data, window, threshold)
        returns = calculate_returns(data, signals)
        
        return {
            'params': params,
            'sharpe': calculate_sharpe(returns),
            'max_dd': calculate_max_drawdown(returns)
        }
    
    # 参数网格
    param_grid = [
        (w, t) 
        for w in range(10, 50, 5) 
        for t in np.arange(0.5, 2.0, 0.1)
    ]
    
    data = load_historical_data()
    
    # 方法1：ProcessPoolExecutor（推荐）
    with ProcessPoolExecutor(max_workers=mp.cpu_count()) as executor:
        results = list(
            executor.map(
                partial(backtest_strategy, data), 
                param_grid
            )
        )
    
    # 找到最优参数
    best_result = max(results, key=lambda x: x['sharpe'])
    print(f"最优参数: {best_result['params']}")
    print(f"Sharpe比率: {best_result['sharpe']:.4f}")
    
    return results

# 场景：多标的回测
def backtest_multiple_symbols_parallel(symbols, start_date, end_date):
    """并行回测多个标的"""
    
    def backtest_single_symbol(symbol):
        """回测单个标的"""
        data = load_data(symbol, start_date, end_date)
        strategy = MyStrategy()
        results = strategy.backtest(data)
        return symbol, results
    
    # 使用多线程（I/O密集）或多进程（CPU密集）
    use_multiprocessing = True  # 根据任务类型选择
    
    if use_multiprocessing:
        with ProcessPoolExecutor() as executor:
            results = list(executor.map(backtest_single_symbol, symbols))
    else:
        with ThreadPoolExecutor() as executor:
            results = list(executor.map(backtest_single_symbol, symbols))
    
    return dict(results)

import pandas as pd
import numpy as np

def optimize_dtypes(df):
    """优化DataFrame数据类型"""
    
    # 原始内存使用
    original_memory = df.memory_usage(deep=True).sum() / 1024**2
    print(f"原始内存: {original_memory:.2f} MB")
    
    # 优化数值列
    for col in df.select_dtypes(include=['int64']).columns:
        col_min = df[col].min()
        col_max = df[col].max()
        
        # 选择最小的可行类型
        if col_min >= 0:
            if col_max < 255:
                df[col] = df[col].astype(np.uint8)
            elif col_max < 65535:
                df[col] = df[col].astype(np.uint16)
            elif col_max < 4294967295:
                df[col] = df[col].astype(np.uint32)
        else:
            if col_min > -128 and col_max < 127:
                df[col] = df[col].astype(np.int8)
            elif col_min > -32768 and col_max < 32767:
                df[col] = df[col].astype(np.int16)
            elif col_min > -2147483648 and col_max < 2147483647:
                df[col] = df[col].astype(np.int32)
    
    # 优化浮点列
    for col in df.select_dtypes(include=['float64']).columns:
        df[col] = pd.to_numeric(df[col], downcast='float')
    
    # 优化类别列
    for col in df.select_dtypes(include=['object']).columns:
        if df[col].nunique() / len(df) < 0.5:  # 唯一值比例小于50%
            df[col] = df[col].astype('category')
    
    # 优化后内存使用
    optimized_memory = df.memory_usage(deep=True).sum() / 1024**2
    print(f"优化后内存: {optimized_memory:.2f} MB")
    print(f"节省: {original_memory - optimized_memory:.2f} MB ({100*(1-optimized_memory/original_memory):.1f}%)")
    
    return df

# 使用示例
df = pd.read_csv('large_dataset.csv')
df_optimized = optimize_dtypes(df)

def process_large_dataset_chunks(file_path, chunk_size=100000):
    """分块处理大型数据集"""
    
    results = []
    
    # 使用生成器逐块读取
    for chunk in pd.read_csv(file_path, chunksize=chunk_size):
        # 处理每个chunk
        processed = process_chunk(chunk)
        results.append(processed)
        
        # 释放内存
        del chunk
    
    # 合并结果
    return pd.concat(results, ignore_index=True)

def process_chunk(df):
    """处理单个数据块"""
    # 计算因子
    df['factor'] = calculate_factor(df)
    
    # 生成信号
    df['signal'] = generate_signal(df['factor'])
    
    return df[['date', 'symbol', 'factor', 'signal']]

import pandas as pd
import pickle
import h5py
import parquet

def benchmark_data_formats():
    """不同数据格式的读写性能对比"""
    
    # 创建测试数据
    np.random.seed(42)
    n_rows = 1000000
    n_cols = 50
    
    df = pd.DataFrame({
        f'col_{i}': np.random.randn(n_rows)
        for i in range(n_cols)
    })
    df['date'] = pd.date_range('2020-01-01', periods=n_rows, freq='T')
    df['symbol'] = np.random.choice(['AAPL', 'GOOGL', 'MSFT'], n_rows)
    
    results = {}
    
    # CSV格式
    start = time.time()
    df.to_csv('data.csv', index=False)
    csv_read_time = time.time() - start
    
    start = time.time()
    df_csv = pd.read_csv('data.csv')
    csv_write_time = time.time() - start
    
    results['CSV'] = {
        'write_time': csv_write_time,
        'read_time': csv_read_time,
        'file_size': os.path.getsize('data.csv') / 1024**2
    }
    
    # Parquet格式（推荐）
    start = time.time()
    df.to_parquet('data.parquet', compression='snappy')
    parquet_write_time = time.time() - start
    
    start = time.time()
    df_parquet = pd.read_parquet('data.parquet')
    parquet_read_time = time.time() - start
    
    results['Parquet'] = {
        'write_time': parquet_write_time,
        'read_time': parquet_read_time,
        'file_size': os.path.getsize('data.parquet') / 1024**2
    }
    
    # HDF5格式
    start = time.time()
    df.to_hdf('data.h5', key='data', mode='w')
    hdf_write_time = time.time() - start
    
    start = time.time()
    df_hdf = pd.read_hdf('data.h5', key='data')
    hdf_read_time = time.time() - start
    
    results['HDF5'] = {
        'write_time': hdf_write_time,
        'read_time': hdf_read_time,
        'file_size': os.path.getsize('data.h5') / 1024**2
    }
    
    # 输出对比结果
    print("\n数据格式性能对比：")
    print("=" * 80)
    for format_name, metrics in results.items():
        print(f"\n{format_name}:")
        print(f"  写入时间: {metrics['write_time']:.4f}秒")
        print(f"  读取时间: {metrics['read_time']:.4f}秒")
        print(f"  文件大小: {metrics['file_size']:.2f} MB")
    
    return results

import sqlite3
import pandas as pd
from sqlalchemy import create_engine

def optimize_database_queries():
    """数据库查询优化示例"""
    
    # 创建数据库连接
    engine = create_engine('sqlite:///trading_data.db')
    
    # 低效查询：逐行读取
    def inefficient_query(symbols):
        """低效的逐符号查询"""
        results = []
        for symbol in symbols:
            query = f"""
            SELECT date, close, volume
            FROM daily_prices
            WHERE symbol = '{symbol}'
            AND date BETWEEN '2020-01-01' AND '2025-12-31'
            """
            df = pd.read_sql(query, engine)
            results.append(df)
        
        return pd.concat(results, ignore_index=True)
    
    # 优化查询：批量读取
    def optimized_query(symbols):
        """优化的批量查询"""
        
        # 方法1：使用IN子句
        symbols_tuple = tuple(symbols)
        query = f"""
        SELECT date, symbol, close, volume
        FROM daily_prices
        WHERE symbol IN {symbols_tuple}
        AND date BETWEEN '2020-01-01' AND '2025-12-31'
        """
        df = pd.read_sql(query, engine)
        
        return df
    
    # 方法2：使用索引
    def query_with_index(symbol):
        """使用索引加速查询"""
        query = f"""
        SELECT /*+ INDEX(daily_prices idx_symbol_date) */
        date, close, volume
        FROM daily_prices
        WHERE symbol = '{symbol}'
        AND date BETWEEN '2020-01-01' AND '2025-12-31'
        """
        return pd.read_sql(query, engine)
    
    # 方法3：预处理和缓存
    def query_with_cache(symbol, cache={}):
        """使用缓存避免重复查询"""
        cache_key = f"{symbol}_2020_2025"
        
        if cache_key not in cache:
            query = f"""
            SELECT date, close, volume
            FROM daily_prices
            WHERE symbol = '{symbol}'
            AND date BETWEEN '2020-01-01' AND '2025-12-31'
            """
            cache[cache_key] = pd.read_sql(query, engine)
        
        return cache[cache_key]

from functools import lru_cache
import pickle
import hashlib

class FactorCache:
    """因子计算缓存系统"""
    
    def __init__(self, cache_dir='./cache'):
        self.cache_dir = cache_dir
        os.makedirs(cache_dir, exist_ok=True)
    
    def get_cache_key(self, func_name, *args, **kwargs):
        """生成缓存key"""
        # 将参数序列化为字符串
        param_str = f"{func_name}_{str(args)}_{str(kwargs)}"
        
        # 计算hash
        cache_key = hashlib.md5(param_str.encode()).hexdigest()
        
        return cache_key
    
    def cached_factor_calculation(self, func):
        """因子计算缓存装饰器"""
        
        def wrapper(*args, **kwargs):
            # 生成缓存key
            cache_key = self.get_cache_key(func.__name__, *args, **kwargs)
            cache_file = os.path.join(self.cache_dir, f"{cache_key}.pkl")
            
            # 检查缓存
            if os.path.exists(cache_file):
                print(f"从缓存加载: {func.__name__}")
                with open(cache_file, 'rb') as f:
                    return pickle.load(f)
            
            # 计算并缓存
            print(f"计算并缓存: {func.__name__}")
            result = func(*args, **kwargs)
            
            with open(cache_file, 'wb') as f:
                pickle.dump(result, f)
            
            return result
        
        return wrapper

# 使用示例
cache = FactorCache()

@cache.cached_factor_calculation
def calculate_complex_factor(data, param1, param2):
    """复杂的因子计算（耗时）"""
    # 模拟耗时计算
    time.sleep(5)
    
    factor = data['close'].rolling(param1).mean() / data['volume'].rolling(param2).mean()
    
    return factor

class LazyDataLoader:
    """懒加载数据加载器"""
    
    def __init__(self, data_source):
        self.data_source = data_source
        self._cache = {}
    
    def __getitem__(self, key):
        """延迟加载数据"""
        if key not in self._cache:
            print(f"加载数据: {key}")
            self._cache[key] = self.load_data(key)
        
        return self._cache[key]
    
    def load_data(self, key):
        """实际的数据加载逻辑"""
        # 这里可以是从数据库、文件、API加载
        if self.data_source == 'csv':
            return pd.read_csv(f'data/{key}.csv')
        elif self.data_source == 'parquet':
            return pd.read_parquet(f'data/{key}.parquet')
        else:
            raise ValueError(f"不支持的数据源: {self.data_source}")

# 使用示例
loader = LazyDataLoader('parquet')

# 只有在访问时才加载
factor_data = loader['factor_2024']  # 这里才触发加载

import time
import psutil
import threading
from collections import deque

class PerformanceMonitor:
    """实时性能监控器"""
    
    def __init__(self, window_size=100):
        self.window_size = window_size
        self.cpu_usage = deque(maxlen=window_size)
        self.memory_usage = deque(maxlen=window_size)
        self.execution_times = deque(maxlen=window_size)
        self.is_monitoring = False
    
    def start_monitoring(self):
        """开始监控"""
        self.is_monitoring = True
        
        def monitor_loop():
            while self.is_monitoring:
                # 记录CPU使用率
                cpu_percent = psutil.cpu_percent(interval=1)
                self.cpu_usage.append(cpu_percent)
                
                # 记录内存使用率
                memory = psutil.virtual_memory()
                self.memory_usage.append(memory.percent)
                
                time.sleep(1)
        
        monitor_thread = threading.Thread(target=monitor_loop)
        monitor_thread.daemon = True
        monitor_thread.start()
    
    def stop_monitoring(self):
        """停止监控"""
        self.is_monitoring = False
    
    def record_execution_time(self, execution_time):
        """记录执行时间"""
        self.execution_times.append(execution_time)
    
    def get_statistics(self):
        """获取性能统计"""
        stats = {
            'avg_cpu': np.mean(self.cpu_usage) if self.cpu_usage else 0,
            'avg_memory': np.mean(self.memory_usage) if self.memory_usage else 0,
            'avg_execution_time': np.mean(self.execution_times) if self.execution_times else 0,
            'max_execution_time': max(self.execution_times) if self.execution_times else 0,
        }
        
        return stats
    
    def print_report(self):
        """打印性能报告"""
        stats = self.get_statistics()
        
        print("\n" + "="*80)
        print("性能监控报告")
        print("="*80)
        print(f"平均CPU使用率: {stats['avg_cpu']:.1f}%")
        print(f"平均内存使用率: {stats['avg_memory']:.1f}%")
        print(f"平均执行时间: {stats['avg_execution_time']:.4f}秒")
        print(f"最长执行时间: {stats['max_execution_time']:.4f}秒")
        print("="*80)

class OptimizedBacktestSystem:
    """优化后的回测系统"""
    
    def __init__(self, data, initial_capital=1000000):
        self.data = data
        self.initial_capital = initial_capital
        self.positions = np.zeros(len(data))
        self.cash = initial_capital
        self.portfolio_value = np.zeros(len(data))
        
        # 预计算常用指标
        self._precompute_indicators()
    
    def _precompute_indicators(self):
        """预计算常用技术指标"""
        # 使用numba加速计算
        self.returns = self.data['close'].pct_change().values
        self.ma20 = calculate_moving_average_numba(self.data['close'].values, 20)
        self.ma60 = calculate_moving_average_numba(self.data['close'].values, 60)
        self.volatility = calculate_volatility_parallel(self.returns, 20)
    
    @njit(parallel=True)
    def generate_signals_vectorized(self):
        """向量化信号生成"""
        n = len(self.data)
        signals = np.zeros(n)
        
        # 并行生成信号
        for i in prange(60, n):
            # 均线金叉
            if self.ma20[i] > self.ma60[i] and self.ma20[i-1] <= self.ma60[i-1]:
                signals[i] = 1
            # 均线死叉
            elif self.ma20[i] < self.ma60[i] and self.ma20[i-1] >= self.ma60[i-1]:
                signals[i] = -1
        
        return signals
    
    def run_backtest(self):
        """运行回测"""
        signals = self.generate_signals_vectorized()
        
        # 向量化仓位计算
        self.positions = signals
        self.portfolio_value = self.cash + self.positions * self.data['close'].values
        
        return self.calculate_performance()
    
    def calculate_performance(self):
        """计算性能指标"""
        # 使用numba加速计算
        returns = np.diff(self.portfolio_value) / self.portfolio_value[:-1]
        
        total_return = (self.portfolio_value[-1] / self.initial_capital - 1) * 100
        sharpe = np.mean(returns) / np.std(returns) * np.sqrt(252)
        max_dd = self.calculate_max_drawdown(self.portfolio_value)
        
        return {
            'total_return': total_return,
            'sharpe_ratio': sharpe,
            'max_drawdown': max_dd
        }
    
    @staticmethod
    @njit
    def calculate_max_drawdown(portfolio_value):
        """计算最大回撤（numba加速）"""
        n = len(portfolio_value)
        peak = portfolio_value[0]
        max_dd = 0.0
        
        for i in range(1, n):
            if portfolio_value[i] > peak:
                peak = portfolio_value[i]
            
            drawdown = (peak - portfolio_value[i]) / peak
            if drawdown > max_dd:
                max_dd = drawdown
        
        return max_dd * 100