Tutorial 02 - Moving averages and MACD through decomposition

from pathlib import Path

import matplotlib.pyplot as plt
import numpy as np
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.classic_indicators import macd, sma
from examples.quant_trading.features import decompose_one_series, walkforward_decompose_ohlcv
from examples.quant_trading.strategy_baselines import make_classic_baseline_weight_grid, run_classical_baselines
from examples.quant_trading.strategy_detime import (
    compare_classical_and_detime,
    make_detime_trend_weight_grid,
    run_detime_trend_baselines,
)
from examples.quant_trading.validation import compare_weight_strategies, write_run_audit

pd.set_option("display.max_columns", 20)
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"]

DECOMP_PERIOD_CANDIDATES = (63, 126, 252)
DECOMP_TRAIN_WINDOW = 504
DECOMP_STEP = 5
DECOMP_Z_WINDOW = 63

features = walkforward_decompose_ohlcv(
    ohlcv,
    method="STL",
    period="auto",
    period_candidates=DECOMP_PERIOD_CANDIDATES,
    train_window=DECOMP_TRAIN_WINDOW,
    step=DECOMP_STEP,
    z_window=DECOMP_Z_WINDOW,
)
prices.tail()

classical_weights = make_classic_baseline_weight_grid(prices)
classical_table, classical_results = compare_weight_strategies(prices, classical_weights, fee_bps=1.0, slippage_bps=2.0)
display(classical_table[["total_return", "cagr", "sharpe", "max_drawdown", "average_turnover"]].round(4))

fig, ax = plt.subplots(figsize=(10, 4))
prices[ticker].plot(ax=ax, linewidth=1.0, label="close")
sma(prices, 20)[ticker].plot(ax=ax, linewidth=1.0, label="SMA 20")
sma(prices, 100)[ticker].plot(ax=ax, linewidth=1.0, label="SMA 100")
ax.set_title("Classical dual moving-average inputs")
ax.legend()
ax.grid(True, alpha=0.25)
plt.show()

image/png

classic_macd = macd(prices, fast=12, slow=26, signal=9)
fig, ax = plt.subplots(figsize=(10, 3))
classic_macd["macd"][ticker].plot(ax=ax, linewidth=1.0, label="MACD")
classic_macd["signal"][ticker].plot(ax=ax, linewidth=1.0, label="signal")
ax.axhline(0, linewidth=0.8)
ax.set_title("Classical MACD on raw close")
ax.legend()
ax.grid(True, alpha=0.25)
plt.show()

image/png

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

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

trend_price = np.exp(diagnostic["trend"]).to_frame(ticker)
fig, ax = plt.subplots(figsize=(10, 4))
prices[ticker].plot(ax=ax, linewidth=0.9, color="#1f2937", label="close")
trend_price[ticker].plot(ax=ax, linewidth=2.0, color="#0f766e", label="continuous DeTime trend")
sma(trend_price, 20)[ticker].plot(ax=ax, linewidth=1.0, color="#f97316", label="trend SMA 20")
sma(trend_price, 100)[ticker].plot(ax=ax, linewidth=1.0, color="#dc2626", label="trend SMA 100")
ax.set_title("Continuous decomposition trend used by MA rewrites")
ax.legend()
ax.grid(True, alpha=0.25)
plt.show()

image/png

diagnostic_trend = diagnostic["trend"].to_frame(ticker)
trend_macd = macd(diagnostic_trend, fast=12, slow=26, signal=9)
fig, axes = plt.subplots(2, 1, figsize=(10, 4.6), sharex=True, gridspec_kw={"height_ratios": [1.2, 1.0]})
trend_macd["macd"][ticker].plot(ax=axes[0], linewidth=1.2, color="#2563eb", label="trend MACD")
trend_macd["signal"][ticker].plot(ax=axes[0], linewidth=1.0, color="#f97316", label="trend signal")
axes[0].axhline(0, color="black", linewidth=0.8)
axes[0].set_title("MACD computed on the continuous decomposed trend")
axes[0].legend(loc="best")
axes[0].grid(True, alpha=0.25)

diagnostic["cycle_z"].plot(ax=axes[1], linewidth=0.9, color="#16a34a", label="cycle z")
axes[1].axhline(0, color="black", linewidth=0.8)
axes[1].set_title("Cycle context used as a timing filter")
axes[1].legend(loc="best")
axes[1].grid(True, alpha=0.25)
plt.tight_layout()
plt.show()

image/png

fig, ax = plt.subplots(figsize=(10, 3))
diagnostic["residual_abs_z"].plot(ax=ax, linewidth=1.1, color="#7c3aed", label="price residual abs z")
volume_diagnostic["residual_z"].plot(ax=ax, linewidth=0.9, color="#0f766e", alpha=0.75, label="volume residual z")
ax.axhline(2.75, linestyle="--", linewidth=0.9, color="#991b1b")
ax.set_title("Continuous residual stress and volume participation")
ax.legend()
ax.grid(True, alpha=0.25)
plt.show()

image/png

comparison = compare_classical_and_detime(
    run_classical_baselines(prices, fee_bps=1.0, slippage_bps=2.0),
    run_detime_trend_baselines(prices, features, fee_bps=1.0, slippage_bps=2.0),
)
display(comparison[["strategy_group", "total_return", "cagr", "sharpe", "max_drawdown", "average_turnover"]].round(4))

all_results = {**classical_results, **detime_results}
leaders = comparison.head(5).index.tolist()
fig, ax = plt.subplots(figsize=(10, 4))
for name in leaders:
    all_results[name].equity.plot(ax=ax, linewidth=1.2, label=name)
ax.set_title("Equity curves: top strategies by Sharpe")
ax.legend(loc="best", fontsize=8)
ax.grid(True, alpha=0.25)
plt.show()

image/png

fig, ax = plt.subplots(figsize=(10, 3))
for name in leaders[:4]:
    dd = all_results[name].equity / all_results[name].equity.cummax() - 1.0
    dd.plot(ax=ax, linewidth=1.0, label=name)
ax.set_title("Drawdown comparison")
ax.legend(loc="best", fontsize=8)
ax.grid(True, alpha=0.25)
plt.show()

image/png

plot_table = comparison.copy()
fig, ax = plt.subplots(figsize=(7, 4))
ax.scatter(plot_table["max_drawdown"].abs(), plot_table["sharpe"], s=60)
for name, row in plot_table.iterrows():
    ax.annotate(name.replace("classic_", "c_").replace("detime_", "d_"), (abs(row["max_drawdown"]), row["sharpe"]), fontsize=7, xytext=(4, 4), textcoords="offset points")
ax.set_xlabel("absolute max drawdown")
ax.set_ylabel("Sharpe")
ax.set_title("Risk-adjusted comparison")
ax.grid(True, alpha=0.25)
plt.show()

image/png

fig, ax = plt.subplots(figsize=(10, 3))
for name in ["classic_sma_20_100", "detime_sma_20_100_trend_volume", "detime_macd_12_26_9"]:
    weights = (classical_weights | detime_weights)[name]
    weights[ticker].plot(ax=ax, drawstyle="steps-post", linewidth=1.0, label=name)
ax.set_title("Position states")
ax.legend(loc="best", fontsize=8)
ax.grid(True, alpha=0.25)
plt.show()

image/png

paths = write_run_audit(
    REPORT_DIR,
    data_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 historical OHLCV sample"),
    ),
    audit=ohlcv_audit_report(ohlcv),
    strategy_stats=comparison,
    prefix="column_02",
)
display(pd.DataFrame({"artifact": list(paths), "path": [str(p) for p in paths.values()]}))