《海龟交易法则》Python量化交易程序实现

一、策略核心规则回顾

1. 入场出场信号（唐奇安通道）

• 短期系统（系统1）：20日突破入场，10日突破止损出场
  • 做多：收盘价突破20日最高价开多；跌破10日最低价平多
  • 做空：收盘价跌破20日最低价开空；突破10日最高价平空
• 长期系统（系统2）：55日突破入场，20日突破止损出场
  • 做多：收盘价突破55日最高价开多；跌破20日最低价平多
  • 做空：收盘价跌破55日最低价开空；突破20日最高价平空

2. 仓位与风险控制（ATR波动率仓位）

1. ATR(20)：20日平均真实波幅，衡量标的波动
2. 单次单笔风险：账户总资金的1%
3. 单位仓位Size计算公式：
   $$Size=\frac{账户资金\times 1\%}{ATR\times 每点价值}$$
4. 单品种最大4单位，全市场合计12单位上限，防止过度重仓

3. 止损规则

开仓后反向N倍ATR止损：系统1止损2ATR，系统2止损2ATR

二、完整Python代码

依赖库：pandas, numpy, yfinance（免费获取A股/美股数据）

import numpy as np
import pandas as pd
import yfinance as yf
from datetime import datetime

class TurtleTradingSystem:
    def __init__(
        self,
        initial_capital: float = 100000,
        risk_pct: float = 0.01,    # 单次风险1%账户资金
        atr_period: int = 20,
        entry_short: int = 20,     # 系统1入场周期
        exit_short: int = 10,      # 系统1出场周期
        entry_long: int = 55,      # 系统2入场周期
        exit_long: int = 20,       # 系统2出场周期
        stop_loss_mult: float = 2.0# 2倍ATR止损
    ):
        # 资金与风险参数
        self.capital = initial_capital
        self.risk_pct = risk_pct
        self.atr_period = atr_period
        self.sl_mult = stop_loss_mult

        # 唐奇安通道周期
        self.entry_s = entry_short
        self.exit_s = exit_short
        self.entry_l = entry_long
        self.exit_l = exit_long

        # 持仓记录
        self.position = 0    # 0无仓 / 正数多单 / 负数空单
        self.entry_price = 0.0
        self.stop_price = 0.0
        self.trade_log = [] # 交易日志

    @staticmethod
    def calc_atr(df: pd.DataFrame, period: int) -> pd.Series:
        """计算ATR平均真实波幅"""
        high = df["High"]
        low = df["Low"]
        close = df["Close"].shift(1)
        tr1 = high - low
        tr2 = abs(high - close)
        tr3 = abs(low - close)
        tr = pd.concat([tr1, tr2, tr3], axis=1).max(axis=1)
        atr = tr.rolling(window=period).mean()
        return atr

    def build_indicator(self, df: pd.DataFrame) -> pd.DataFrame:
        """计算全部海龟指标：ATR、唐奇安高低轨"""
        data = df.copy()
        # ATR
        data["ATR"] = self.calc_atr(data, self.atr_period)

        # 系统1 唐奇安通道
        data["high20"] = data["High"].rolling(self.entry_s).max()
        data["low20"] = data["Low"].rolling(self.entry_s).min()
        data["high10"] = data["High"].rolling(self.exit_s).max()
        data["low10"] = data["Low"].rolling(self.exit_s).min()

        # 系统2 唐奇安通道
        data["high55"] = data["High"].rolling(self.entry_l).max()
        data["low55"] = data["Low"].rolling(self.entry_l).min()
        data["high20_exit"] = data["High"].rolling(self.exit_l).max()
        data["low20_exit"] = data["Low"].rolling(self.exit_l).min()
        return data.dropna()

    def calc_position_size(self, atr: float, tick_value: float = 1) -> int:
        """计算可开仓单位数量
        tick_value：每波动1点对应盈亏，股票填1，期货填合约乘数
        """
        risk_dollar = self.capital * self.risk_pct
        unit_risk = atr * tick_value * self.sl_mult
        size = risk_dollar / unit_risk
        return int(np.floor(size))

    def open_long(self, price: float, atr: float, tick_value: float):
        """开多仓，设置止损"""
        size = self.calc_position_size(atr, tick_value)
        self.position = size
        self.entry_price = price
        self.stop_price = price - self.sl_mult * atr
        print(f"【开多】价格:{price:.2f} 仓位:{size} 止损:{self.stop_price:.2f}")

    def open_short(self, price: float, atr: float, tick_value: float):
        """开空仓，设置止损"""
        size = self.calc_position_size(atr, tick_value)
        self.position = -size
        self.entry_price = price
        self.stop_price = price + self.sl_mult * atr
        print(f"【开空】价格:{price:.2f} 仓位:{-size} 止损:{self.stop_price:.2f}")

    def close_position(self, price: float, date):
        """平仓，更新账户资金+记录交易"""
        profit = (price - self.entry_price) * self.position
        self.capital += profit
        self.trade_log.append({
            "date": date,
            "entry_price": self.entry_price,
            "exit_price": price,
            "position": self.position,
            "profit": profit,
            "capital_after": self.capital
        })
        print(f"【平仓】价格:{price:.2f} 盈亏:{profit:.2f} 剩余资金:{self.capital:.2f}")
        self.position = 0
        self.entry_price = 0
        self.stop_price = 0

    def run_backtest(self, df: pd.DataFrame, tick_value: float = 1, use_system2: bool = True):
        """回测主逻辑
        use_system2=True 同时运行系统1+系统2；False仅系统1
        """
        data = self.build_indicator(df)
        for idx, row in data.iterrows():
            close = row["Close"]
            atr = row["ATR"]
            date = idx

            # 持仓状态：检查止损
            if self.position != 0:
                # 多单止损跌破止损价平仓
                if self.position > 0 and close <= self.stop_price:
                    self.close_position(close, date)
                    continue
                # 空单止损突破止损价平仓
                if self.position < 0 and close >= self.stop_price:
                    self.close_position(close, date)
                    continue

            # 无持仓，判断入场信号
            if self.position == 0:
                # 系统1信号
                long_signal_s = close > row["high20"]
                short_signal_s = close < row["low20"]
                # 系统2信号
                long_signal_l = close > row["high55"]
                short_signal_l = close < row["low55"]

                # 优先执行信号（二选一做多做空，不同时开）
                if long_signal_s or (use_system2 and long_signal_l):
                    self.open_long(close, atr, tick_value)
                elif short_signal_s or (use_system2 and short_signal_l):
                    self.open_short(close, atr, tick_value)
            else:
                # 持有仓位时，检查通道出场信号（海龟原版双重出场：止损+通道突破）
                if self.position > 0:
                    # 多单通道出场：跌破10/20日线低点
                    exit_s = close < row["low10"]
                    exit_l = close < row["low20_exit"]
                    if exit_s or (use_system2 and exit_l):
                        self.close_position(close, date)
                elif self.position < 0:
                    # 空单通道出场：突破10/20日线高点
                    exit_s = close > row["high10"]
                    exit_l = close > row["high20_exit"]
                    if exit_s or (use_system2 and exit_l):
                        self.close_position(close, date)

        # 回测结束输出总结
        trade_df = pd.DataFrame(self.trade_log)
        print("\n====== 回测总结 ======")
        print(f"初始资金: 100000")
        print(f"期末资金: {self.capital:.2f}")
        print(f"总收益率: {(self.capital/100000-1)*100:.2f}%")
        print(f"总交易次数: {len(trade_df)}")
        if len(trade_df) > 0:
            print(f"总盈亏: {trade_df['profit'].sum():.2f}")
        return trade_df

# ---------------------- 测试运行 ----------------------
if __name__ == "__main__":
    # 1. 下载标的数据，示例：苹果股票AAPL，可替换A股代码如600030.SS
    ticker = yf.Ticker("AAPL")
    df = ticker.history(start="2015-01-01", end="2026-01-01")
    df = df[["Open", "High", "Low", "Close", "Volume"]]

    # 2. 初始化海龟交易系统
    turtle = TurtleTradingSystem(initial_capital=100000, risk_pct=0.01)

    # 3. 执行回测 tick_value=1代表股票1元/1股，期货修改为合约乘数
    trade_result = turtle.run_backtest(df, tick_value=1, use_system2=True)

    # 4. 输出交易明细
    print("\n全部交易记录：")
    print(trade_result.round(2))

三、代码关键说明

1. 数据接口

• yfinance免费获取美股/港股/沪深A股日线数据
• A股代码规则：沪市xxxxxx.SS，深市xxxxxx.SZ，例如600519.SS贵州茅台

2. 参数适配期货/股票

• 股票：tick_value = 1，单位为1股
• 期货（如沪深300股指）：tick_value=300，波动1点盈亏300元

3. 原版海龟双重出场机制

1. 硬止损：开仓后固定2ATR止损，极端行情强制离场
2. 通道出场：反向突破唐奇安通道出场，捕捉趋势结束信号

4. 风险控制

严格遵循原版1%账户单笔风险，自动根据波动率调整仓位：

• 高波动品种 → ATR大 → 开仓数量减少
• 低波动品种 → ATR小 → 可加仓放大仓位

四、实盘改造扩展点（重要）

1. 多品种并行交易

增加标的循环，加入总仓位上限12单位控制，多品种同时持仓时合计仓位不超限：

# 全局总仓位计数器
total_units = 0
max_total_units = 12
# 开仓前判断
if abs(new_size) + total_units > max_total_units:
    new_size = max_total_units - total_units

2. 滑点与手续费

在平仓盈亏中扣除交易成本，贴近实盘：

commission = 0.0003  # 万三手续费
slippage = 0.01      # 1分钱滑点
profit = (price - self.entry_price) * self.position - abs(self.position)*(commission + slippage)

3. 过滤假突破（原版海龟过滤规则）

增加55日趋势过滤：仅价格高于55日均线只做多、低于只做空，减少震荡亏损：

data["ma55"] = data["Close"].rolling(55).mean()
trend_up = row["Close"] > row["ma55"]
trend_down = row["Close"] < row["ma55"]
# 做多仅在trend_up时触发
if long_signal_s and trend_up:
    self.open_long(...)

4. 对接实盘接口

• A股：聚宽、米筐、同花顺iFinQ API
• 美股：IB盈透、Interactive Brokers Python API
• 期货：CTP期货交易接口

五、回测局限性提示

1. 本代码为日线回测，原版海龟使用日线级别趋势跟踪，不适合短线
2. 趋势跟踪策略在长期震荡行情会连续小额止损，适合牛市/大单边行情
3. 未加入最大回撤、夏普比率、胜率等绩效指标，可自行补充绩效分析模块
4. 历史回测收益不代表未来实盘表现，需多品种、多周期样本验证

六、绩效指标扩展代码（可选追加）

在回测结束后增加风险指标计算：

def calc_performance(trade_df, initial_cap):
    if len(trade_df) == 0:
        return {}
    cap_series = [initial_cap]
    for p in trade_df["profit"]:
        cap_series.append(cap_series[-1] + p)
    cap_series = np.array(cap_series)
    returns = np.diff(cap_series) / cap_series[:-1]
    total_return = cap_series[-1]/initial_cap - 1
    max_drawdown = np.max(np.maximum.accumulate(cap_series) - cap_series) / np.max(np.maximum.accumulate(cap_series))
    win_trades = trade_df[trade_df["profit"]>0]
    win_rate = len(win_trades)/len(trade_df)
    return {
        "总收益率": f"{total_return*100:.2f}%",
        "最大回撤": f"{max_drawdown*100:.2f}%",
        "胜率": f"{win_rate*100:.2f}%",
        "交易次数": len(trade_df)
    }

请用 tushare 替换 yfinance

---

基于Tushare的海龟交易法则完整Python回测代码

替换原yfinance，使用tushare获取A股日线数据，适配国内股票规则，保留原版海龟完整逻辑（双系统、ATR仓位、双重出场、1%风控）

前置依赖安装

pip install tushare pandas numpy

完整代码

import numpy as np
import pandas as pd
import tushare as ts

# ===================== 1. 配置Tushare Token =====================
# 去tushare官网个人中心获取token，替换此处
TUSHARE_TOKEN = "你的Tusharetoken"
ts.set_token(TUSHARE_TOKEN)
pro = ts.pro_api()

class TurtleTradingSystem:
    def __init__(
        self,
        initial_capital: float = 1000000,
        risk_pct: float = 0.01,    # 单笔风险占总资金1%
        atr_period: int = 20,
        entry_short: int = 20,     # 系统1入场20日突破
        exit_short: int = 10,      # 系统1出场10日通道
        entry_long: int = 55,      # 系统2入场55日突破
        exit_long: int = 20,       # 系统2出场20日通道
        stop_loss_mult: float = 2.0# 2倍ATR止损
    ):
        self.capital = initial_capital
        self.risk_pct = risk_pct
        self.atr_period = atr_period
        self.sl_mult = stop_loss_mult

        self.entry_s = entry_short
        self.exit_s = exit_short
        self.entry_l = entry_long
        self.exit_l = exit_long

        # 持仓状态
        self.position = 0
        self.entry_price = 0.0
        self.stop_price = 0.0
        self.trade_log = []

    @staticmethod
    def calc_atr(df: pd.DataFrame, period: int) -> pd.Series:
        """计算ATR平均真实波幅"""
        high = df["high"]
        low = df["low"]
        close_prev = df["close"].shift(1)
        tr1 = high - low
        tr2 = abs(high - close_prev)
        tr3 = abs(low - close_prev)
        tr = pd.concat([tr1, tr2, tr3], axis=1).max(axis=1)
        atr = tr.rolling(window=period).mean()
        return atr

    def build_indicator(self, df: pd.DataFrame) -> pd.DataFrame:
        """计算唐奇安通道+ATR"""
        data = df.copy()
        data["ATR"] = self.calc_atr(data, self.atr_period)

        # 系统1通道
        data["high20"] = data["high"].rolling(self.entry_s).max()
        data["low20"] = data["low"].rolling(self.entry_s).min()
        data["high10"] = data["high"].rolling(self.exit_s).max()
        data["low10"] = data["low"].rolling(self.exit_s).min()

        # 系统2通道
        data["high55"] = data["high"].rolling(self.entry_l).max()
        data["low55"] = data["low"].rolling(self.entry_l).min()
        data["high20_exit"] = data["high"].rolling(self.exit_l).max()
        data["low20_exit"] = data["low"].rolling(self.exit_l).min()

        return data.dropna()

    def calc_position_size(self, atr: float, tick_value: float = 1) -> int:
        """计算开仓股数，A股tick_value=1（1股）"""
        risk_total = self.capital * self.risk_pct
        single_risk = atr * tick_value * self.sl_mult
        size = risk_total / single_risk
        return int(np.floor(size))

    def open_long(self, price: float, atr: float, tick_value: float):
        """开多仓，设置止损"""
        size = self.calc_position_size(atr, tick_value)
        self.position = size
        self.entry_price = price
        self.stop_price = price - self.sl_mult * atr
        print(f"【开多】价格:{price:.2f} 仓位:{size}股 止损:{self.stop_price:.2f}")

    def open_short(self, price: float, atr: float, tick_value: float):
        """开空仓（A股仅融券支持做空，回测保留逻辑）"""
        size = self.calc_position_size(atr, tick_value)
        self.position = -size
        self.entry_price = price
        self.stop_price = price + self.sl_mult * atr
        print(f"【开空】价格:{price:.2f} 仓位:{-size}股 止损:{self.stop_price:.2f}")

    def close_position(self, price: float, trade_date):
        """平仓，更新资金+记录交易"""
        profit = (price - self.entry_price) * self.position
        # 模拟万3手续费
        fee_rate = 0.0003
        fee = abs(self.position) * self.entry_price * fee_rate
        profit = profit - fee
        self.capital += profit

        self.trade_log.append({
            "date": trade_date,
            "entry_price": round(self.entry_price, 2),
            "exit_price": round(price, 2),
            "shares": self.position,
            "profit": round(profit, 2),
            "capital_after": round(self.capital, 2)
        })
        print(f"【平仓】价格:{price:.2f} 净盈亏:{profit:.2f} 账户资金:{self.capital:.2f}")
        self.position = 0
        self.entry_price = 0
        self.stop_price = 0

    def run_backtest(self, df: pd.DataFrame, tick_value: float = 1, use_system2: bool = True):
        """回测主逻辑"""
        data = self.build_indicator(df)
        for idx, row in data.iterrows():
            close = row["close"]
            atr = row["ATR"]
            dt = idx

            # 持仓优先检查硬止损
            if self.position != 0:
                if self.position > 0 and close <= self.stop_price:
                    self.close_position(close, dt)
                    continue
                if self.position < 0 and close >= self.stop_price:
                    self.close_position(close, dt)
                    continue

            # 无持仓，判断突破入场
            if self.position == 0:
                sig_long_s = close > row["high20"]
                sig_short_s = close < row["low20"]
                sig_long_l = close > row["high55"]
                sig_short_l = close < row["low55"]

                if sig_long_s or (use_system2 and sig_long_l):
                    self.open_long(close, atr, tick_value)
                elif sig_short_s or (use_system2 and sig_short_l):
                    self.open_short(close, atr, tick_value)
            else:
                # 通道出场信号
                if self.position > 0:
                    exit_s = close < row["low10"]
                    exit_l = close < row["low20_exit"]
                    if exit_s or (use_system2 and exit_l):
                        self.close_position(close, dt)
                elif self.position < 0:
                    exit_s = close > row["high10"]
                    exit_l = close > row["high20_exit"]
                    if exit_s or (use_system2 and exit_l):
                        self.close_position(close, dt)

        trade_df = pd.DataFrame(self.trade_log)
        print("\n========== 回测总结 ==========")
        print(f"初始资金: 100000.00")
        print(f"期末资金: {self.capital:.2f}")
        ret = (self.capital / 100000 - 1) * 100
        print(f"总收益率: {ret:.2f}%")
        print(f"总交易次数: {len(trade_df)}")
        if len(trade_df) > 0:
            print(f"累计盈亏: {trade_df['profit'].sum():.2f}")
        return trade_df

# ===================== 工具函数：Tushare获取A股日线 =====================
def get_tushare_daily(stock_code: str, start_date: str, end_date: str):
    """
    获取A股日线行情
    stock_code: 股票代码 如 600519.SH / 000001.SZ
    """
    df = pro.daily(ts_code=stock_code, start_date=start_date, end_date=end_date)
    # 日期转时间索引并升序排列
    df["trade_date"] = pd.to_datetime(df["trade_date"])
    df = df.sort_values("trade_date").reset_index(drop=True)
    df = df.set_index("trade_date")
    # 保留海龟需要字段
    df = df[["open", "high", "low", "close", "vol"]]
    return df

# ===================== 回测入口 =====================
if __name__ == "__main__":
    # 1. 参数配置
    stock = "300308.SZ"   # 中际旭创
    start = "20150101"
    end = "20260710"

    # 2. 拉取行情数据
    print("正在从Tushare下载行情数据...")
    price_df = get_tushare_daily(stock, start, end)
    print(f"数据长度: {len(price_df)} 根K线")

    # 3. 初始化海龟系统并回测
    turtle = TurtleTradingSystem(initial_capital=1000000)
    trade_records = turtle.run_backtest(price_df, tick_value=1, use_system2=True)

    # 打印全部交易记录
    print("\n全部交易明细：")
    print(trade_records)

关键修改说明（Tushare适配点）

1. 数据源替换

   • 移除yfinance，使用tushare.pro_api()拉取A股日线
   • Tushare字段名：open/high/low/close/vol（小写），和美股yfinance大小写区分，代码全部适配小写列名
   • 日期格式：Tushare返回trade_date为YYYYMMDD字符串，统一转datetime时间索引

2. A股交易适配

   • tick_value=1：A股1股对应1元波动，股票无需修改；期货需调整乘数
   • 内置万三手续费模拟，平仓自动扣减交易成本
   • 做空逻辑保留，但A股普通账户无法裸空，仅融券可做空，实盘做多为主

3. Tushare使用步骤

   1. 官网注册：https://tushare.pro/
   2. 个人主页获取Token，替换代码中TUSHARE_TOKEN = "你的Tusharetoken"
   3. 积分要求：daily日线接口需要基础积分，注册后完成任务即可获取

4. 股票代码规则

   • 沪市：xxxxxx.SH 60开头
   • 深市主板/中小板：00xxxx.SZ
   • 创业板：30xxxx.SZ
   • 科创板：688xxx.SH

可选：增加绩效指标计算函数

在代码末尾追加，回测完成后调用：

def calc_performance_metrics(trade_df, init_cap):
    if len(trade_df) == 0:
        return "无交易记录"
    cap_list = [init_cap]
    for p in trade_df["profit"]:
        cap_list.append(cap_list[-1] + p)
    cap_arr = np.array(cap_list)
    # 总收益
    total_return = cap_arr[-1] / init_cap - 1
    # 最大回撤
    peak = np.maximum.accumulate(cap_arr)
    drawdown = (peak - cap_arr) / peak
    max_dd = drawdown.max()
    # 胜率
    win_trades = trade_df[trade_df["profit"] > 0]
    win_rate = len(win_trades) / len(trade_df)
    res = {
        "总收益率": f"{total_return*100:.2f}%",
        "最大回撤": f"{max_dd*100:.2f}%",
        "交易胜率": f"{win_rate*100:.2f}%",
        "盈利次数": len(win_trades),
        "亏损次数": len(trade_df) - len(win_trades)
    }
    return res

# 使用示例，放在run_backtest之后
metrics = calc_performance_metrics(trade_records, 100000)
print("\n绩效指标：", metrics)

实盘拓展提示

1. 多品种并行：循环多只股票，增加全局总单位上限12控制，原版海龟限制总仓位
2. 涨跌停处理：Tushare数据包含真实涨跌停，可增加过滤跳过无法开仓K线
3. 实盘接口：对接聚宽/米筐/掘金量化，将信号生成逻辑单独抽离下单模块
4. 过滤优化：增加55日均线趋势过滤，只顺势开仓，减少震荡连续止损