Tutorial 04 - Donchian and Turtle breakout with DeTime volume confirmation

from pathlib import Path

import matplotlib.pyplot as plt
import pandas as pd
from IPython.display import display

from examples.quant_trading.data import load_sample_goog_ohlcv, market_data_manifest, ohlcv_audit_report
from examples.quant_trading.features import build_feature_table, decompose_one_series, walkforward_decompose_ohlcv
from examples.quant_trading.classic_indicators import donchian_channels
from examples.quant_trading.strategy_baselines import make_classic_breakout_weight_grid, run_classical_breakout_baselines
from examples.quant_trading.strategy_detime import make_detime_breakout_weight_grid, run_detime_breakout_baselines, compare_classical_and_detime
from examples.quant_trading.validation import compare_weight_strategies, write_run_audit, write_run_manifest

pd.set_option("display.max_columns", 30)
report_dir = Path("examples/quant_trading/reports")
report_dir.mkdir(parents=True, exist_ok=True)

ohlcv_single = load_sample_goog_ohlcv(trim_start="2014-01-01")
ticker = ohlcv_single.attrs.get("symbol", "GOOG")
ohlcv = {field: ohlcv_single[[field]].rename(columns={field: ticker}) for field in ["Open", "High", "Low", "Close", "Volume"]}
prices = ohlcv["Close"]
high = ohlcv["High"]
low = ohlcv["Low"]
manifest = market_data_manifest(tickers=[ticker], start=str(prices.index.min().date()), end=str(prices.index.max().date()), interval="1d", source=ohlcv_single.attrs.get("source", "bundled real sample"))
audit = ohlcv_audit_report(ohlcv)
display(audit)

classic_breakout_weights = make_classic_breakout_weight_grid(prices, high=high, low=low)
classic_table, classic_results = compare_weight_strategies(prices, classic_breakout_weights, fee_bps=1.0, slippage_bps=2.0)
display(classic_table[["total_return", "cagr", "sharpe", "max_drawdown", "average_turnover"]].round(4))

channel = donchian_channels(high, low, window=55, shift=1)
fig, ax = plt.subplots(figsize=(10, 4))
prices[ticker].plot(ax=ax, linewidth=1.0, label="close")
channel.upper[ticker].plot(ax=ax, linewidth=0.9, label="55-day upper channel")
channel.lower[ticker].plot(ax=ax, linewidth=0.9, label="55-day lower channel")
ax.set_title("Classical Donchian channel")
ax.legend()
ax.grid(True, alpha=0.3)
plt.show()

image/png

features = walkforward_decompose_ohlcv(
    ohlcv,
    method="STL",
    period="auto",
    period_candidates=(63, 126, 252),
    train_window=504,
    step=5,
    z_window=63,
)
feature_tail = build_feature_table(prices, features).tail(120)
display(feature_tail.tail(5).round(4))
feature_tail.to_csv(report_dir / "column_04_feature_table_tail.csv")

diagnostic = decompose_one_series(
    prices[ticker],
    method="STL",
    period="auto",
    period_candidates=(63, 126, 252),
    z_window=63,
    transform="log",
)
volume_diagnostic = decompose_one_series(
    ohlcv["Volume"][ticker],
    method="STL",
    period=int(diagnostic.attrs.get("period", 126)),
    z_window=63,
    transform="log1p",
)

fig, ax = plt.subplots(figsize=(10, 3.8))
diagnostic["trend_strength"].plot(ax=ax, linewidth=1.0, color="#0f766e", label="trend strength")
diagnostic["cycle_z"].plot(ax=ax, linewidth=0.9, color="#16a34a", label="cycle z")
diagnostic["residual_abs_z"].plot(ax=ax, linewidth=0.9, color="#7c3aed", alpha=0.85, label="absolute residual z")
volume_diagnostic["residual_z"].plot(ax=ax, linewidth=0.8, color="#f97316", alpha=0.75, label="volume residual z")
ax.axhline(0, color="black", linewidth=0.8)
ax.set_title("Continuous structural context used by DeTime breakout")
ax.legend()
ax.grid(True, alpha=0.3)
plt.show()

image/png

detime_breakout_weights = make_detime_breakout_weight_grid(prices, features, high=high, low=low)
detime_table, detime_results = compare_weight_strategies(prices, detime_breakout_weights, fee_bps=1.0, slippage_bps=2.0)
display(detime_table[["total_return", "cagr", "sharpe", "max_drawdown", "average_turnover"]].round(4))

classical = run_classical_breakout_baselines(prices, high=high, low=low, fee_bps=1.0, slippage_bps=2.0)
detime = run_detime_breakout_baselines(prices, features, high=high, low=low, fee_bps=1.0, slippage_bps=2.0)
comparison = compare_classical_and_detime(classical, detime)
display(comparison[["strategy_group", "cagr", "sharpe", "max_drawdown", "average_turnover", "hit_rate"]].round(4))
comparison.to_csv(report_dir / "column_04_strategy_comparison.csv")
write_run_audit(report_dir, data_manifest=manifest, audit=audit, strategy_stats=comparison, prefix="column_04")
manifest_path = write_run_manifest(
    report_dir / "column_04_run_manifest.json",
    command="notebook:04_turtle_donchian_breakout_volume_confirmation",
    dataset="bundled_real_GOOG",
    strategies=list(comparison.index),
    result_file=str(report_dir / "column_04_strategy_comparison.csv"),
)
manifest_path.as_posix()

'examples/quant_trading/reports/column_04_run_manifest.json'

chosen = "detime_donchian_55_20_volume"
fig, ax1 = plt.subplots(figsize=(10, 4))
prices[ticker].plot(ax=ax1, linewidth=1.0, label="close")
channel.upper[ticker].plot(ax=ax1, linewidth=0.8, label="55-day upper")
ax1.set_title(f"Position overlay: {chosen}")
ax2 = ax1.twinx()
detime_breakout_weights[chosen][ticker].plot(ax=ax2, linewidth=0.9, alpha=0.7, label="position")
ax2.set_ylabel("position")
ax1.grid(True, alpha=0.3)
plt.show()

image/png