极客工具箱
随手即取的小型效率工具集合。纯前端计算,数据不上传。
📚 预设模板:
📥 专栏01: 数据获取
📊 专栏02: 收益率与止损
🔜 专栏03: 移动平均线
🔜 专栏04: 量价突破
股票代码:
开始:
结束:
应用参数
🐍 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)
📊 执行结果
点击「运行代码」查看结果