vectorbt-expert

VectorBT 回测专家技能

环境

• Python，搭配 vectorbt、pandas、numpy、plotly
• 数据源：OpenAlgo（印度市场）、DuckDB（直接访问数据库）、yfinance（美国／全球）、CCXT（加密货币）、自定义提供方
• DuckDB 支持：支持自定义 DuckDB 和 OpenAlgo Historify 格式
• API 密钥通过 python-dotenv 和 find_dotenv() 从根目录的单个 .env 文件加载 — 切勿硬编码密钥
• 技术指标：OpenAlgo ta（默认 — from openalgo import ta，涵盖趋势、动量、波动率、成交量、震荡指标、统计和混合指标，共 100 多种）。仅当用户明确要求使用 TA-Lib/talib 时才使用 TA-Lib。无论使用哪种库，绝不使用 VectorBT 内置指标。
• 专用指标（无对应的 TA-Lib 指标，始终使用 openalgo.ta）：Supertrend、Donchian、Ichimoku、HMA、KAMA、ALMA、ZLEMA、VWMA
• 信号清理：使用 openalgo.ta 处理 exrem、crossover、crossunder、flip（始终如此，无论使用哪种指标库）
• 费用模型：印度市场标准（STT + 法定费用 + 每笔订单 Rs 20）
• 基准：默认通过 OpenAlgo 获取 NIFTY 50（NSE_INDEX）
• 图表：Plotly，使用 template="plotly_dark"
• 环境变量通过 find_dotenv() 从项目根目录的单个 .env 文件加载（从脚本目录向上查找）
• 脚本放在 backtesting/{strategy_name}/ 目录中（按需创建，不预先创建）
• 代码或日志输出中绝不使用图标／表情符号

关键规则

1. 默认使用 OpenAlgo ta（from openalgo import ta）处理所有技术指标（EMA、SMA、RSI、MACD、BBANDS、ATR、ADX、STDDEV、MOM 等 90 多种）。仅当用户在提示词中明确要求“talib”／“TA-Lib”时才使用 TA-Lib。绝不使用 vbt.MA.run()、vbt.RSI.run() 或任何 VectorBT 内置指标，无论搭配哪种库。
2. 对于 TA-Lib 中完全没有的指标，始终使用 OpenAlgo ta：Supertrend、Donchian、Ichimoku、HMA、KAMA、ALMA、ZLEMA、VWMA — 这些指标没有对应的 TA-Lib 指标，因此即使是在选择使用 TA-Lib 的脚本中，也应使用 openalgo.ta。
3. 使用 OpenAlgo ta 处理信号工具：ta.exrem()、ta.crossover()、ta.crossunder()、ta.flip()。如果无法导入 openalgo.ta（独立使用 DuckDB 时），则使用内联 exrem() 备用实现。参见 duckdb-data。
4. 生成原始买入／卖出信号后，始终使用 ta.exrem() 清理信号。调用 exrem 前始终使用 .fillna(False)。
5. 按市场选择费用：印度（indian-market-costs）、美国（us-market-costs）、加密货币（crypto-market-costs）。根据用户的市场自动选择。
6. 默认基准：印度=NIFTY（通过 OpenAlgo 获取），美国=S&P 500（^GSPC），加密货币=Bitcoin（BTC-USD）。参见 data-fetching 中的市场选择指南。
7. 每次回测后，始终提供策略与基准的对比表。
8. 始终用通俗易懂的语言解读回测报告，让普通交易者也能理解风险和优势。
9. Plotly K 线图必须设置 xaxis type="category"，以避免出现周末间隔。
10. 整股交易：股票始终设置 min_size=1, size_granularity=1。
11. DuckDB 数据加载：用户提供 DuckDB 路径时，使用 duckdb.connect() 和 read_only=True 直接加载数据。自动检测格式：OpenAlgo Historify（表 market_data，纪元时间戳）或自定义格式（表 ohlcv，包含日期和时间列）。参见 duckdb-data。

模块化规则文件

各主题的详细参考资料见 rules/：

规则文件	主题
data-fetching	OpenAlgo（印度）、yfinance（美国）、CCXT（加密货币）、自定义提供方、.env 配置
simulation-modes	from_signals、from_orders、from_holding、方向类型
position-sizing	Amount/Value/Percent/TargetPercent 仓位规模设置
indicators-signals	OpenAlgo ta 指标参考（默认）、可选启用 TA-Lib、信号生成
openalgo-ta-helpers	完整的 OpenAlgo ta 目录（100 多种指标）：exrem、crossover、Supertrend、Donchian、Ichimoku、均线
stop-loss-take-profit	固定止损、止盈、跟踪止损
parameter-optimization	广播和基于循环的优化
performance-analysis	统计数据、指标、基准比较、CAGR
plotting	K 线图（类别 x 轴）、VectorBT 图表、自定义 Plotly
indian-market-costs	按细分市场划分的印度市场费用模型
us-market-costs	美国市场费用模型（股票、期权、期货）
crypto-market-costs	加密货币费用模型（现货、USDT-M、COIN-M 期货）
futures-backtesting	手数（SEBI 于 2025 年 12 月修订）、价值规模设置
long-short-trading	同时进行多头/空头交易、方向比较
duckdb-data	DuckDB 直接加载、Historify 格式、自动检测、重采样、多标的
csv-data-resampling	加载 CSV、按印度市场时间对齐进行重采样
walk-forward	滚动前向分析、WFE 比率
robustness-testing	Monte Carlo、噪声测试、参数敏感性、延迟测试
pitfalls	常见错误和上线前检查清单
strategy-catalog	策略参考及代码片段
openstatz-tearsheet	OpenStatz 交互式离线仪表板、指标、Monte Carlo（替代 QuantStats）

策略模板（位于 rules/assets/）

可用于生产环境的脚本，包含真实费用、NIFTY 基准、对比表和通俗易懂的报告：

模板	路径	描述
EMA 交叉	assets/ema_crossover/backtest.py	EMA 10/20 交叉
RSI	assets/rsi/backtest.py	RSI(14) 超卖/超买
Donchian	assets/donchian/backtest.py	Donchian 通道突破
Supertrend	assets/supertrend/backtest.py	带日内交易时段的 Supertrend
MACD	assets/macd/backtest.py	MACD 信号 K 线突破
SDA2	assets/sda2/backtest.py	SDA2 趋势跟踪
动量	assets/momentum/backtest.py	双重动量（MOM + MOM-of-MOM）
双重动量	assets/dual_momentum/backtest.py	按季度轮换 ETF
买入并持有	assets/buy_hold/backtest.py	静态多资产配置
RSI 累积	assets/rsi_accumulation/backtest.py	按周以 RSI 分层累积
滚动前向	assets/walk_forward/template.py	滚动前向分析模板
真实费用	assets/realistic_costs/template.py	交易成本影响比较

快速模板：标准回测脚本

import os
from datetime import datetime, timedelta
from pathlib import Path

import numpy as np
import pandas as pd
import vectorbt as vbt
from dotenv import find_dotenv, load_dotenv
from openalgo import api, ta

# --- Config ---
script_dir = Path(__file__).resolve().parent
load_dotenv(find_dotenv(), override=False)

SYMBOL = "SBIN"
EXCHANGE = "NSE"
INTERVAL = "D"
INIT_CASH = 1_000_000
FEES = 0.00111              # Indian delivery equity (STT + statutory)
FIXED_FEES = 20             # Rs 20 per order
ALLOCATION = 0.75
BENCHMARK_SYMBOL = "NIFTY"
BENCHMARK_EXCHANGE = "NSE_INDEX"

# --- Fetch Data ---
client = api(
    api_key=os.getenv("OPENALGO_API_KEY"),
    host=os.getenv("OPENALGO_HOST", "http://127.0.0.1:5000"),
)

end_date = datetime.now().date()
start_date = end_date - timedelta(days=365 * 3)

df = client.history(
    symbol=SYMBOL, exchange=EXCHANGE, interval=INTERVAL,
    start_date=start_date.strftime("%Y-%m-%d"),
    end_date=end_date.strftime("%Y-%m-%d"),
)
if "timestamp" in df.columns:
    df["timestamp"] = pd.to_datetime(df["timestamp"])
    df = df.set_index("timestamp")
else:
    df.index = pd.to_datetime(df.index)
df = df.sort_index()
if df.index.tz is not None:
    df.index = df.index.tz_convert(None)

close = df["close"]

# --- Strategy: EMA Crossover (OpenAlgo ta - default indicator library) ---
ema_fast = ta.ema(close, 10)
ema_slow = ta.ema(close, 20)

buy_raw = (ema_fast > ema_slow) & (ema_fast.shift(1) <= ema_slow.shift(1))
sell_raw = (ema_fast < ema_slow) & (ema_fast.shift(1) >= ema_slow.shift(1))

entries = ta.exrem(buy_raw.fillna(False), sell_raw.fillna(False))
exits = ta.exrem(sell_raw.fillna(False), buy_raw.fillna(False))

# --- Backtest ---
pf = vbt.Portfolio.from_signals(
    close, entries, exits,
    init_cash=INIT_CASH, size=ALLOCATION, size_type="percent",
    fees=FEES, fixed_fees=FIXED_FEES, direction="longonly",
    min_size=1, size_granularity=1, freq="1D",
)

# --- Benchmark ---
df_bench = client.history(
    symbol=BENCHMARK_SYMBOL, exchange=BENCHMARK_EXCHANGE, interval=INTERVAL,
    start_date=start_date.strftime("%Y-%m-%d"),
    end_date=end_date.strftime("%Y-%m-%d"),
)
if "timestamp" in df_bench.columns:
    df_bench["timestamp"] = pd.to_datetime(df_bench["timestamp"])
    df_bench = df_bench.set_index("timestamp")
else:
    df_bench.index = pd.to_datetime(df_bench.index)
df_bench = df_bench.sort_index()
if df_bench.index.tz is not None:
    df_bench.index = df_bench.index.tz_convert(None)
bench_close = df_bench["close"].reindex(close.index).ffill().bfill()
pf_bench = vbt.Portfolio.from_holding(bench_close, init_cash=INIT_CASH, fees=FEES, freq="1D")

# --- Results ---
print(pf.stats())

# --- Strategy vs Benchmark ---
comparison = pd.DataFrame({
    "Strategy": [
        f"{pf.total_return() * 100:.2f}%", f"{pf.sharpe_ratio():.2f}",
        f"{pf.sortino_ratio():.2f}", f"{pf.max_drawdown() * 100:.2f}%",
        f"{pf.trades.win_rate() * 100:.1f}%", f"{pf.trades.count()}",
        f"{pf.trades.profit_factor():.2f}",
    ],
    f"Benchmark ({BENCHMARK_SYMBOL})": [
        f"{pf_bench.total_return() * 100:.2f}%", f"{pf_bench.sharpe_ratio():.2f}",
        f"{pf_bench.sortino_ratio():.2f}", f"{pf_bench.max_drawdown() * 100:.2f}%",
        "-", "-", "-",
    ],
}, index=["Total Return", "Sharpe Ratio", "Sortino Ratio", "Max Drawdown",
          "Win Rate", "Total Trades", "Profit Factor"])
print(comparison.to_string())

# --- Explain ---
print(f"* Total Return: {pf.total_return() * 100:.2f}% vs NIFTY {pf_bench.total_return() * 100:.2f}%")
print(f"* Max Drawdown: {pf.max_drawdown() * 100:.2f}%")
print(f"  -> On Rs {INIT_CASH:,}, worst temporary loss = Rs {abs(pf.max_drawdown()) * INIT_CASH:,.0f}")

# --- Plot ---
fig = pf.plot(subplots=['value', 'underwater', 'cum_returns'], template="plotly_dark")
fig.show()

# --- Export ---
pf.positions.records_readable.to_csv(script_dir / f"{SYMBOL}_trades.csv", index=False)

快速模板：DuckDB 回测脚本

import datetime as dt
from pathlib import Path

import duckdb
import numpy as np
import pandas as pd
import vectorbt as vbt

try:
    # Default: OpenAlgo ta for both indicators and signal cleaning
    from openalgo import ta
    exrem = ta.exrem
    ema = ta.ema
except ImportError:
    # Fallback ONLY when the openalgo package itself is not installed
    # (standalone DuckDB with no OpenAlgo). Use TA-Lib for indicators
    # and this inline exrem() replacement for signal cleaning.
    import talib as tl

    def ema(data, period):
        return pd.Series(tl.EMA(data.values, timeperiod=period), index=data.index)

    def exrem(signal1, signal2):
        result = signal1.copy()
        active = False
        for i in range(len(signal1)):
            if active:
                result.iloc[i] = False
            if signal1.iloc[i] and not active:
                active = True
            if signal2.iloc[i]:
                active = False
        return result

# --- Config ---
SYMBOL = "SBIN"
DB_PATH = r"path/to/market_data.duckdb"
INIT_CASH = 1_000_000
FEES = 0.000225              # Intraday equity
FIXED_FEES = 20

# --- Load from DuckDB ---
con = duckdb.connect(DB_PATH, read_only=True)
df = con.execute("""
    SELECT date, time, open, high, low, close, volume
    FROM ohlcv WHERE symbol = ? ORDER BY date, time
""", [SYMBOL]).fetchdf()
con.close()

df["datetime"] = pd.to_datetime(df["date"].astype(str) + " " + df["time"].astype(str))
df = df.set_index("datetime").sort_index()
df = df.drop(columns=["date", "time"])

# --- Resample to 5min ---
df_5m = df.resample("5min", origin="start_day", offset="9h15min",
                     label="right", closed="right").agg({
    "open": "first", "high": "max", "low": "min", "close": "last", "volume": "sum"
}).dropna()
close = df_5m["close"]

# --- Strategy + Backtest (same as OpenAlgo template, but use the ema()/exrem() resolved above) ---

如果用户明确要求使用 TA-Lib，请跳过上面的 try/except，直接执行 import talib as tl；exrem 回退方案仅适用于 openalgo 本身不可用的情况。