【量化专栏 02】让数据开口说话：收益率计算与绝对止损的代码实现

import akshare as ak
import pandas as pd

def process_trading_signals(symbol: str, start_date: str, end_date: str) -> pd.DataFrame:
    """
    获取数据并计算收益率与止损信号
    """
    print(f"正在拉取并处理 {symbol} 的行情数据...")
    
    df = ak.stock_zh_a_hist(symbol=symbol, period="daily", start_date=start_date, end_date=end_date, adjust="qfq")
    
    # 基础数据校验
    if df is None or df.empty:
        raise ValueError(f"数据获取失败！请检查网络或股票代码：{symbol}")
    
    df = df.rename(columns={'日期': 'date', '开盘': 'open', '收盘': 'close', '最高': 'high', '最低': 'low'})
    df['date'] = pd.to_datetime(df['date'])
    df.set_index('date', inplace=True)
    
    # ---------------- 核心计算逻辑 ----------------
    
    # 1. 计算当日收益率 (pct_change 默认计算与前一行的百分比变化)
    df['daily_return'] = df['close'].pct_change()
    
    # 2. 规避未来函数：A股是T+1，如果你今天尾盘买入，你的真实收益其实是明天的涨跌幅。
    # 我们使用 shift(-1) 将明天的收益率向上移动一行，作为"今天买入、明天卖出"的预期收益。
    df['next_day_return'] = df['daily_return'].shift(-1)
    
    # 3. 核心风控：构建 -3% 绝对止损逻辑
    # 假设我们昨日收盘买入，今天的盘中最低价如果跌破昨日收盘价的 3%，即触发止损。
    # 公式：(今日最低价 - 昨日收盘价) / 昨日收盘价 <= -0.03
    df['intraday_drawdown'] = (df['low'] - df['close'].shift(1)) / df['close'].shift(1)
    
    # 生成止损触发信号 (True 代表触发，False 代表安全)
    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']]

# === 测试环节 ===
if __name__ == "__main__":
    # 以 TCL科技(000100) 近期数据为例
    df_tcl = process_trading_signals("000100", "20231001", "20231031")
    
    print("\n--- 收益率与止损信号表 ---")
    
    # 筛选出触发了 -3% 止损线的交易日进行观察
    stop_loss_days = df_tcl[df_tcl['trigger_stop_loss'] == True]
    
    print(f"\n在回测区间内，共有 {len(stop_loss_days)} 天盘中触及 -3% 止损线。")
    print(stop_loss_days[['close', 'low', 'intraday_drawdown', 'trigger_stop_loss']].head())