量化交易 - 惯性聚合

博客园 - 嘉禾世兴

OpenShift+ArgoCD+Azure Devops PyTorch 循环神经网络（RNN） PyTorch 卷积神经网络 PyTorch神经网络 PyTorch基础 Metplotlib库集成学习算法 K 近邻算法支持向量机算法决策树算法逻辑回归算法线性回归算法机器学习算法机器学习 kafka 阿里云Clouder 使用MaxKB和Ollama搭建一个自己的AI大模型 Linux-Prometheus Linux-公有云架构

量化交易

嘉禾世兴 · 2025-08-18 · via 博客园 - 嘉禾世兴

量化交易（Quantitative Trading）也可以理解为高频交易，是利用数学模型、统计学方法和计算机编程来进行金融交易的一种策略。

移动平均交叉策略

import requests
import pandas as pd
import matplotlib.pyplot as plt
import ssl

def get_stock_data_eastmoney(stock_code, start_date, end_date):
    """
    Fetch historical stock data from Eastmoney.com
    
    Parameters:
    stock_code (str): Stock code, e.g. '002475'
    start_date (str): Start date in format 'YYYY-MM-DD'
    end_date (str): End date in format 'YYYY-MM-DD'
    
    Returns:
    DataFrame: DataFrame containing historical stock data
    """
    # Eastmoney API
    url = f"http://push2his.eastmoney.com/api/qt/stock/kline/get"
    
    # Handle stock code prefix
    if stock_code.startswith(('6', '5')):
        market_prefix = '1.'  # Shanghai market
    else:
        market_prefix = '0.'  # Shenzhen market
    secid = f"{market_prefix}{stock_code}"
    
    # Request parameters
    params = {
        "secid": secid,
        "ut": "fa5fd1943c7b386f1734de82369f10d",
        "fields1": "f1,f2,f3,f4,f5,f6",
        "fields2": "f51,f52,f53,f54,f55,f56,f57,f58,f59,f60,f61",
        "klt": "101",  # Daily K-line
        "fqt": "0",  # No adjustment
        "beg": start_date.replace('-', ''),
        "end": end_date.replace('-', '')
    }
    
    # Set request headers
    headers = {
        "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36"
    }
    
    try:
        # Bypass SSL verification (handle corporate network issues)
        ssl._create_default_https_context = ssl._create_unverified_context
        
        # Send request
        response = requests.get(url, params=params, headers=headers)
        response.raise_for_status()
        data = response.json()
        
        if data.get('data') and 'klines' in data['data']:
            # Parse K-line data
            klines = data['data']['klines']
            if not klines:
                print(f"Retrieved K-line data is empty")
                return None
            
            # Create DataFrame
            df = pd.DataFrame([kline.split(',')[:7] for kline in klines],
                             columns=['Date', 'Open', 'Close', 'High', 'Low', 'Volume', 'Turnover'])
            
            # Convert data types
            numeric_columns = ['Open', 'Close', 'High', 'Low', 'Volume', 'Turnover']
            df[numeric_columns] = df[numeric_columns].apply(pd.to_numeric)
            
            # Convert date column to datetime type
            df['Date'] = pd.to_datetime(df['Date'])
            
            # Set date as index
            df.set_index('Date', inplace=True)
            
            return df
        else:
            print(f"Failed to retrieve data for stock {stock_code}: {data.get('msg', 'Unknown error')}")
            return None
            
    except requests.exceptions.RequestException as e:
        print(f"Request exception for stock {stock_code}: {e}")
        return None
    except Exception as e:
        print(f"Error processing stock {stock_code}: {e}")
        return None

# Get Luxshare stock data
symbol = "002475"
start_date = "2020-01-01"
end_date = "2025-08-18"

print(f"Fetching stock {symbol} data from {start_date} to {end_date}...")
data = get_stock_data_eastmoney(symbol, start_date, end_date)

if data is not None and not data.empty:
    print("Data retrieved successfully!")
    
    # Use 'Close' column for calculations
    close_column = 'Close'
    
    # Calculate short-term (50-day) and long-term (200-day) moving averages
    data['MA_50'] = data[close_column].rolling(window=50).mean()
    data['MA_200'] = data[close_column].rolling(window=200).mean()

    # Generate buy/sell signals
    data['Signal'] = 0
    data['Signal'][data['MA_50'] > data['MA_200']] = 1   # Short-term MA crosses above long-term MA, generate buy signal
    data['Signal'][data['MA_50'] < data['MA_200']] = -1  # Short-term MA crosses below long-term MA, generate sell signal

    # To better display buy/sell signals, find signal change points
    data['Position'] = data['Signal'].diff()
    
    # Plot stock price and moving averages
    plt.figure(figsize=(14, 8))
    plt.plot(data[close_column], label='Close Price')
    plt.plot(data['MA_50'], label='50-day Moving Average')
    plt.plot(data['MA_200'], label='200-day Moving Average')

    # Mark buy/sell signals
    buy_signals = data[data['Position'] == 1]
    sell_signals = data[data['Position'] == -1]
    
    plt.scatter(buy_signals.index, buy_signals['MA_50'], marker='^', color='g', s=100, label='Buy Signal', zorder=5)
    plt.scatter(sell_signals.index, sell_signals['MA_50'], marker='v', color='r', s=100, label='Sell Signal', zorder=5)

    plt.title(f"Luxshare ({symbol}) Stock Price with Moving Average Strategy")
    plt.xlabel("Date")
    plt.ylabel("Price (CNY)")
    plt.legend()
    plt.grid(True, alpha=0.3)
    plt.tight_layout()
    plt.show()
    
    # Display basic data statistics
    print("\nBasic Data Statistics:")
    print(data[[close_column, 'MA_50', 'MA_200', 'Signal']].describe())
    
    # Display recent buy/sell signals
    print("\nRecent Buy/Sell Signals:")
    recent_signals = data[data['Position'] != 0].tail(10)
    for date, row in recent_signals.iterrows():
        signal_type = "Buy" if row['Position'] == 1 else "Sell"
        print(f"{date.date()}: {signal_type} Signal - Price: {row[close_column]:.2f}")
        
    # Display current status
    if not data.empty:
        latest = data.iloc[-1]
        print(f"\nCurrent Status:")
        print(f"Current Price: {latest[close_column]:.2f}")
        print(f"50-day MA: {latest['MA_50']:.2f}")
        print(f"200-day MA: {latest['MA_200']:.2f}")
        print(f"Current Signal: {'Hold' if latest['Signal'] == 1 else 'No Position'}")
        
else:
    print("Failed to retrieve valid data")

这里计算出40是买点，感觉也不大准啊

量化交易基本策略

量化交易基于数学模型、统计学和计算机算法，通过系统性的方法识别和执行交易机会。

以下是一些常见的量化交易基本策略：

量化交易基于数学模型、统计学和计算机算法，通过系统性的方法识别和执行交易机会。

以下是一些常见的量化交易基本策略：

1. 移动平均策略

策略思想： 基于股价的历史平均值，通过计算短期和长期移动平均线之间的差异来产生买卖信号。
实现方式： 计算短期和长期移动平均，当短期均线上穿长期均线时产生买入信号，反之产生卖出信号。

2. 均值回归策略

策略思想： 基于价格的历史波动，认为价格在波动后会回归到其平均水平。
实现方式： 通过计算价格与均值之间的差异，当价格偏离均值过多时产生买入或卖出信号。

3. 动量策略

策略思想： 基于价格的趋势，认为价格趋势会延续一段时间。
实现方式： 通过计算价格的变化率或其他趋势指标，产生买入或卖出信号。

4. 市场中性策略

策略思想： 通过同时进行买入和卖出，以利用市场的相对强弱。
实现方式： 基于两个或多个相关资产之间的价差或相关性，产生交易信号。

5. 统计套利策略

策略思想： 基于统计学原理，寻找价格之间的临时不平衡，以实现套利。
实现方式： 通过寻找价格、波动性或其他统计指标的异常值，产生交易信号。

6. 事件驱动策略

策略思想： 基于特定事件或信息的发生，产生交易信号。
实现方式： 监控新闻、财报、经济指标等，当发生特定事件时执行交易。

7. 机器学习策略

策略思想： 利用机器学习算法从大量数据中学习模式，预测未来价格走势。
实现方式： 使用回归、分类或深度学习模型，训练模型预测市场走势。

8. 高频交易策略

策略思想： 通过快速执行大量交易来利用极短时间内的小价差。
实现方式： 使用高性能算法和快速执行系统，通常涉及大量交易和低持仓时间。

中国 A 股的股票代码需要加上交易所的后缀，上海证券交易所（SSE）的后缀是 .SS，深圳证券交易所（SZSE）的后缀是 .SZ。

量化回测：

通过回测，交易者可以了解其策略在不同市场条件下的表现，并进行优化和改进。

import requests
import pandas as pd
import matplotlib.pyplot as plt
import ssl

def get_stock_data_eastmoney(stock_code, start_date, end_date):
    """
    Fetch historical stock data from Eastmoney.com
    
    Parameters:
    stock_code (str): Stock code, e.g. '002475'
    start_date (str): Start date in format 'YYYY-MM-DD'
    end_date (str): End date in format 'YYYY-MM-DD'
    
    Returns:
    DataFrame: DataFrame containing historical stock data
    """
    # Eastmoney API
    url = f"http://push2his.eastmoney.com/api/qt/stock/kline/get"
    
    # Handle stock code prefix
    if stock_code.startswith(('6', '5')):
        market_prefix = '1.'  # Shanghai market
    else:
        market_prefix = '0.'  # Shenzhen market
    secid = f"{market_prefix}{stock_code}"
    
    # Request parameters
    params = {
        "secid": secid,
        "ut": "fa5fd1943c7b386f1734de82369f10d",
        "fields1": "f1,f2,f3,f4,f5,f6",
        "fields2": "f51,f52,f53,f54,f55,f56,f57,f58,f59,f60,f61",
        "klt": "101",  # Daily K-line
        "fqt": "0",  # No adjustment
        "beg": start_date.replace('-', ''),
        "end": end_date.replace('-', '')
    }
    
    # Set request headers
    headers = {
        "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36"
    }
    
    try:
        # Bypass SSL verification (handle corporate network issues)
        ssl._create_default_https_context = ssl._create_unverified_context
        
        # Send request
        response = requests.get(url, params=params, headers=headers)
        response.raise_for_status()
        data = response.json()
        
        if data.get('data') and 'klines' in data['data']:
            # Parse K-line data
            klines = data['data']['klines']
            if not klines:
                print(f"Retrieved K-line data is empty")
                return None
            
            # Create DataFrame
            df = pd.DataFrame([kline.split(',')[:7] for kline in klines],
                             columns=['Date', 'Open', 'Close', 'High', 'Low', 'Volume', 'Turnover'])
            
            # Convert data types
            numeric_columns = ['Open', 'Close', 'High', 'Low', 'Volume', 'Turnover']
            df[numeric_columns] = df[numeric_columns].apply(pd.to_numeric)
            
            # Convert date column to datetime type
            df['Date'] = pd.to_datetime(df['Date'])
            
            # Set date as index
            df.set_index('Date', inplace=True)
            
            return df
        else:
            print(f"Failed to retrieve data for stock {stock_code}: {data.get('msg', 'Unknown error')}")
            return None
            
    except requests.exceptions.RequestException as e:
        print(f"Request exception for stock {stock_code}: {e}")
        return None
    except Exception as e:
        print(f"Error processing stock {stock_code}: {e}")
        return None

# Get Luxshare stock data for detailed analysis
symbol = "002475"
start_date = "2022-01-01"
end_date = "2025-08-18"

print(f"Fetching stock {symbol} data from {start_date} to {end_date} for detailed analysis...")
data = get_stock_data_eastmoney(symbol, start_date, end_date)

if data is not None and not data.empty:
    print("Data retrieved successfully! Performing detailed analysis...")
    
    # Use 'Close' column for calculations
    close_column = 'Close'
    
    # Calculate moving averages
    data['SMA_50'] = data[close_column].rolling(window=50).mean()
    data['SMA_200'] = data[close_column].rolling(window=200).mean()

    # Initialize signal column
    data['Signal'] = 0

    # Calculate crossover signals
    data.loc[data['SMA_50'] > data['SMA_200'], 'Signal'] = 1
    data.loc[data['SMA_50'] < data['SMA_200'], 'Signal'] = -1

    # Calculate daily returns
    data['Daily_Return'] = data[close_column].pct_change()

    # Calculate strategy returns (shift(1) to avoid look-ahead bias)
    data['Strategy_Return'] = data['Signal'].shift(1) * data['Daily_Return']

    # Calculate cumulative returns
    data['Cumulative_Return'] = (1 + data['Strategy_Return']).cumprod()
    
    # Calculate buy and hold cumulative returns for comparison
    data['Buy_Hold_Return'] = (1 + data['Daily_Return']).cumprod()

    # Find trade signals
    data['Position'] = data['Signal'].diff()
    
    # Detailed analysis
    print("=== Detailed Strategy Analysis ===")
    
    # Price change analysis
    initial_price = data[close_column].iloc[0]
    final_price = data[close_column].iloc[-1]
    price_change = (final_price / initial_price) - 1
    
    print(f"Initial Price: {initial_price:.2f}")
    print(f"Final Price: {final_price:.2f}")
    print(f"Price Change: {price_change:.2%}")
    
    # Strategy performance
    strategy_return = data['Cumulative_Return'].iloc[-1] - 1
    buy_hold_return = data['Buy_Hold_Return'].iloc[-1] - 1
    
    print(f"\nStrategy Return: {strategy_return:.2%}")
    print(f"Buy & Hold Return: {buy_hold_return:.2%}")
    
    # Signal timing analysis
    print(f"\n=== Signal Timing Analysis ===")
    signals = data[data['Position'] != 0].copy()
    if len(signals) > 0:
        print("Recent Trade Signals:")
        for date, row in signals.tail(10).iterrows():
            signal_type = "BUY" if row['Position'] == 1 else "SELL"
            print(f"{date.date()}: {signal_type} at price {row[close_column]:.2f}")
    
    # Check if we're currently holding
    current_signal = data['Signal'].iloc[-1]
    print(f"\nCurrent Position: {'HOLDING' if current_signal == 1 else 'NOT HOLDING'}")
    
    # Period by period analysis
    print(f"\n=== Key Period Analysis ===")
    
    # Find when strategy was holding vs not holding
    holding_periods = data[data['Signal'] == 1]
    not_holding_periods = data[data['Signal'] == -1]
    
    if len(holding_periods) > 0:
        holding_return = (holding_periods['Daily_Return'] + 1).prod() - 1
        print(f"Return during holding periods: {holding_return:.2%}")
        print(f"Total holding days: {len(holding_periods)}")
    
    if len(not_holding_periods) > 0:
        not_holding_return = (not_holding_periods['Daily_Return'] + 1).prod() - 1
        print(f"Return during non-holding periods: {not_holding_return:.2%}")
        print(f"Total non-holding days: {len(not_holding_periods)}")
    
    # Drawdown analysis
    data['Strategy_Drawdown'] = (data['Cumulative_Return'] / data['Cumulative_Return'].cummax()) - 1
    data['Buy_Hold_Drawdown'] = (data['Buy_Hold_Return'] / data['Buy_Hold_Return'].cummax()) - 1
    
    max_strategy_drawdown = data['Strategy_Drawdown'].min()
    max_buy_hold_drawdown = data['Buy_Hold_Drawdown'].min()
    
    print(f"\n=== Drawdown Analysis ===")
    print(f"Maximum Strategy Drawdown: {max_strategy_drawdown:.2%}")
    print(f"Maximum Buy & Hold Drawdown: {max_buy_hold_drawdown:.2%}")
    
    # Plot detailed chart
    fig, (ax1, ax2) = plt.subplots(2, 1, figsize=(14, 10))
    
    # Price and signals plot
    ax1.plot(data.index, data[close_column], label='Close Price', linewidth=1)
    ax1.plot(data.index, data['SMA_50'], label='50-day MA', alpha=0.7)
    ax1.plot(data.index, data['SMA_200'], label='200-day MA', alpha=0.7)
    
    # Mark buy/sell signals
    buy_signals = data[data['Position'] == 1]
    sell_signals = data[data['Position'] == -1]
    
    ax1.scatter(buy_signals.index, buy_signals[close_column], 
               marker='^', color='green', s=100, label='Buy Signal', zorder=5)
    ax1.scatter(sell_signals.index, sell_signals[close_column], 
               marker='v', color='red', s=100, label='Sell Signal', zorder=5)
    
    ax1.set_title(f'{symbol} Price and Trading Signals')
    ax1.set_ylabel('Price')
    ax1.legend()
    ax1.grid(True, alpha=0.3)
    
    # Returns comparison plot
    ax2.plot(data.index, data['Cumulative_Return'], label='Strategy Return', color='blue')
    ax2.plot(data.index, data['Buy_Hold_Return'], label='Buy & Hold Return', color='green')
    
    ax2.set_title('Cumulative Returns Comparison')
    ax2.set_ylabel('Cumulative Return')
    ax2.set_xlabel('Date')
    ax2.legend()
    ax2.grid(True, alpha=0.3)
    
    plt.tight_layout()
    plt.show()
    
else:
    print("Failed to retrieve valid data for analysis")

根据日均线策略，结果回测收益亏的一塌糊涂。

此内容由惯性聚合(RSS阅读器)自动聚合整理，仅供阅读参考。原文来自 — 版权归原作者所有。

推荐订阅源

博客园 - 嘉禾世兴

移动平均交叉策略

量化交易基本策略

1. 移动平均策略

2. 均值回归策略

3. 动量策略

4. 市场中性策略

5. 统计套利策略

6. 事件驱动策略

7. 机器学习策略

8. 高频交易策略