塔姆量化 | TAM QUANT

🐍 Python 代码

import akshare as ak
import pandas as pd
import warnings
warnings.filterwarnings('ignore')

def process_trading_signals(symbol: str, start_date: str, end_date: str, use_mock: bool = False) -> pd.DataFrame:
    print(f"正在拉取 {symbol} 从 {start_date} 到 {end_date} 的数据...")
    
    if use_mock:
        # 使用模拟数据（网络故障时）
        print("⚠️ 网络获取失败，使用模拟数据...")
        dates = pd.date_range(start='2023-10-01', end='2023-10-31', freq='B')
        import numpy as np
        np.random.seed(42)
        base = 10.0
        data = {
            'open': base + np.random.randn(len(dates)) * 0.5,
            'close': base + np.random.randn(len(dates)) * 0.5,
            'low': base + np.random.randn(len(dates)) * 0.6,
            'high': base + np.random.randn(len(dates)) * 0.6,
        }
        df = pd.DataFrame(data, index=dates)
        df.index.name = 'date'
    else:
        try:
            df = ak.stock_zh_a_hist(symbol=symbol, period="daily", start_date=start_date, end_date=end_date, adjust="qfq", timeout=10)
            if df is None or df.empty:
                raise ValueError("数据为空")
        except Exception as e:
            print(f"⚠️ 数据获取异常: {e}")
            print("自动切换到模拟数据模式...")
            return process_trading_signals(symbol, start_date, end_date, use_mock=True)
    
    df = df.rename(columns={'日期': 'date', '开盘': 'open', '收盘': 'close', '最高': 'high', '最低': 'low'})
    df['date'] = pd.to_datetime(df['date'])
    df.set_index('date', inplace=True)
    
    # 1. 计算当日收益率
    df['daily_return'] = df['close'].pct_change()
    
    # 2. 规避未来函数：T+1 明日收益
    df['next_day_return'] = df['daily_return'].shift(-1)
    
    # 3. 核心风控：-3% 绝对止损
    df['intraday_drawdown'] = (df['low'] - df['close'].shift(1)) / df['close'].shift(1)
    df['trigger_stop_loss'] = df['intraday_drawdown'] <= -0.03
    df = df.dropna()
    
    return df[['open', 'close', 'low', 'daily_return', 'next_day_return', 'intraday_drawdown', 'trigger_stop_loss']]

# === 执行 ===
symbol = "000100"
start_date = "20231001"
end_date = "20231031"

df = process_trading_signals(symbol, start_date, end_date)

print("=" * 60)
print(f"【{symbol} 收益率与止损信号分析】")
print(f"回测区间: {start_date} ~ {end_date}")
print("=" * 60)

# 统计止损触发天数
stop_loss_days = df[df['trigger_stop_loss'] == True]
print(f"\n🚨 触发了 -3% 止损线的交易日: {len(stop_loss_days)} 天")

print("\n【前10条数据】")
print(df[['close', 'daily_return', 'intraday_drawdown', 'trigger_stop_loss']].head(10).to_string())

print("\n【收益率统计】")
print(f"平均日收益率: {df['daily_return'].mean()*100:.2f}%")
print(f"最大单日涨幅: {df['daily_return'].max()*100:.2f}%")
print(f"最大单日跌幅: {df['daily_return'].min()*100:.2f}%")
print("=" * 60)

📊 执行结果

点击「运行代码」查看结果

极客工具箱

🐍 Python 代码

📊 执行结果