Coverage for book/marimo/notebooks/Experiment5.py: 100%
100 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-07-31 10:05 +0000
1# /// script
2# requires-python = ">=3.12"
3# dependencies = [
4# "marimo==0.23.15",
5# "numpy==2.4.6",
6# "plotly==6.9.0",
7# "polars==1.43.1",
8# "jquantstats==0.9.7",
9# "tinycta==0.14.0"
10# ]
11# ///
13"""Experiment 5: Advanced CTA strategy with correlation-based optimization.
15This module implements a sophisticated trend-following strategy that
16incorporates dynamic conditional correlation (DCC) and matrix optimization
17techniques to improve portfolio construction and risk management.
18"""
20import marimo
22__generated_with = "0.23.9"
23app = marimo.App()
25with app.setup:
26 import sys
27 from pathlib import Path
29 import marimo as mo
30 import numpy as np
31 import polars as pl
32 from jquantstats import Portfolio
33 from tinycta.linalg import inv_a_norm, solve
34 from tinycta.osc import osc
35 from tinycta.signal import shrink2id
36 from tinycta.util import vol_adj
38 sys.path.insert(0, str(Path(__file__).parent))
40 from preamble import date_col, load_prices
42 prices = load_prices(__file__)
43 prices_only = prices.drop(date_col)
44 assets = prices_only.columns
47@app.cell(hide_code=True)
48def _():
49 mo.md(r"""
50 # CTA 5.0 - Optimization 2.0
51 """)
52 return
55@app.function
56def f(price: "pl.Expr", fast: int = 32, slow: int = 96, vola: int = 32, clip: float = 4.2) -> "pl.Expr":
57 """Return the tanh oscillator of vol-adjusted cumulative price."""
58 return osc(vol_adj(price, vola=vola, clip=clip, min_samples=300).cum_sum(), fast=fast, slow=slow).tanh()
61@app.cell
62def _():
63 fast = mo.ui.slider(4, 192, step=4, value=32, label="Fast Moving Average")
64 slow = mo.ui.slider(4, 192, step=4, value=96, label="Slow Moving Average")
65 vola = mo.ui.slider(4, 192, step=4, value=32, label="Volatility")
66 winsor = mo.ui.slider(1.0, 6.0, step=0.1, value=4.2, label="Winsorizing")
67 corr = mo.ui.slider(50, 500, step=10, value=200, label="Correlation")
68 shrinkage = mo.ui.slider(0.0, 1.0, step=0.05, value=0.5, label="Shrinkage")
70 mo.vstack([fast, slow, vola, winsor, corr, shrinkage])
71 return corr, shrinkage, vola, winsor
74@app.function
75def ewm_covariance(returns_adj: "pl.DataFrame", *, corr: int) -> "np.ndarray":
76 """EWM covariance tensor of the vol-adjusted returns (the Engle-DCC numerator).
78 ``cov_t(i, j) = ewm_t(r_i * r_j) - ewm_t(r_i) * ewm_t(r_j)``, evaluated for
79 every day; the result has shape ``(n_rows, n_assets, n_assets)`` and is
80 symmetric in ``(i, j)``.
81 """
82 columns = returns_adj.columns
83 n_assets = len(columns)
84 n_rows = len(returns_adj)
86 ewm_means_np = returns_adj.select(pl.all().ewm_mean(com=corr, min_samples=int(corr))).to_numpy()
88 pair_indices = [(i, j) for i in range(n_assets) for j in range(i, n_assets)]
89 ewm_prod_np = returns_adj.select(
90 [
91 (pl.col(columns[i]) * pl.col(columns[j]))
92 .fill_nan(None)
93 .ewm_mean(com=corr, min_samples=int(corr))
94 .alias(f"p{i}_{j}")
95 for i, j in pair_indices
96 ]
97 ).to_numpy()
99 cov_np = np.full((n_rows, n_assets, n_assets), np.nan)
100 for _k, (_i, _j) in enumerate(pair_indices):
101 _cov = ewm_prod_np[:, _k] - ewm_means_np[:, _i] * ewm_means_np[:, _j]
102 cov_np[:, _i, _j] = _cov
103 cov_np[:, _j, _i] = _cov
104 return cov_np
107@app.function
108def correlation_from_covariance(cov_np: "np.ndarray") -> "np.ndarray":
109 """Normalize a per-day covariance tensor to a correlation tensor.
111 Days with non-positive variance keep their NaN off-diagonals; the diagonal
112 is forced to 1 wherever the variance is positive.
113 """
114 n_assets = cov_np.shape[1]
115 _var = cov_np[:, np.arange(n_assets), np.arange(n_assets)]
116 with np.errstate(invalid="ignore", divide="ignore"):
117 _denom = np.sqrt(_var[:, :, None] * _var[:, None, :])
118 cor_3d: np.ndarray = cov_np / _denom
119 for _k in range(n_assets):
120 cor_3d[_var[:, _k] > 0, _k, _k] = 1.0
121 return cor_3d
124@app.function
125def dcc_correlation(prices_only: "pl.DataFrame", *, vola: int, clip: float, corr: int) -> "np.ndarray":
126 """Engle-DCC per-day correlation tensor, shape ``(n_rows, n_assets, n_assets)``."""
127 returns_adj = prices_only.select(vol_adj(pl.all(), vola=vola, clip=clip, min_samples=300))
128 cov_np = ewm_covariance(returns_adj, corr=corr)
129 return correlation_from_covariance(cov_np)
132@app.function
133def positions(
134 cor_3d: "np.ndarray", mu: "np.ndarray", vo: "np.ndarray", prices_np: "np.ndarray", *, shrinkage: float
135) -> "np.ndarray":
136 """Per-day risk-parity positions from the shrunk DCC correlation tensor.
138 Each day, the correlation matrix is shrunk towards the identity, restricted
139 to the assets with a finite price, and used to solve for the risk-scaled
140 position. Days with no live assets or a singular/zero norm are left flat.
141 """
142 n_rows, n_assets = prices_np.shape
143 pos_matrix = np.zeros((n_rows, n_assets))
144 for _n in range(n_rows):
145 _mask = np.isfinite(prices_np[_n])
146 if _mask.sum() == 0:
147 continue
148 _full_shrunk = shrink2id(cor_3d[_n], lamb=shrinkage)
149 _matrix = _full_shrunk[_mask, :][:, _mask]
150 _expected_mu = np.nan_to_num(mu[_n][_mask])
151 _expected_vo = np.nan_to_num(vo[_n][_mask])
152 _norm = inv_a_norm(_expected_mu, _matrix)
153 if _norm == 0 or np.isnan(_norm):
154 continue
155 _risk_pos = solve(_matrix, _expected_mu) / _norm
156 pos_matrix[_n, _mask] = np.nan_to_num(1e6 * _risk_pos / _expected_vo, nan=0.0)
157 return pos_matrix
160@app.cell
161def _(corr, shrinkage, vola, winsor):
162 # EWM correlation (DCC by Engle), then per-day risk-parity positions.
163 cor_3d = dcc_correlation(prices_only, vola=vola.value, clip=winsor.value, corr=corr.value)
165 mu = prices_only.select(f(pl.all(), fast=32, slow=96, vola=vola.value, clip=winsor.value)).to_numpy()
166 vo = prices_only.select(
167 pl.all().fill_nan(None).pct_change().ewm_std(com=vola.value, min_samples=int(vola.value))
168 ).to_numpy()
170 pos_matrix = positions(cor_3d, mu, vo, prices_only.to_numpy(), shrinkage=shrinkage.value)
172 portfolio = Portfolio.from_cash_position(
173 prices=prices,
174 cash_position=pl.concat(
175 [prices.select(date_col), pl.from_numpy(pos_matrix, schema=dict.fromkeys(assets, pl.Float64))],
176 how="horizontal_extend",
177 ),
178 aum=1e8,
179 )
180 return (portfolio,)
183@app.cell
184def _(portfolio):
185 print(portfolio.stats.sharpe())
186 return
189@app.cell(hide_code=True)
190def _():
191 mo.md(r"""
192 # Conclusions
193 * Dramatic relative improvements observable despite using the same signals as in previous Experiment.
194 * Main difference here is to take advantage of cross-correlations in the risk measurement.
195 * Possible to add constraints on individual assets or groups of them.
196 * Possible to reflect trading costs in objective with regularization terms (Ridge, Lars, Elastic Nets, ...)
197 """)
198 return
201@app.cell
202def _(portfolio):
203 portfolio.plots.snapshot()
204 return
207@app.cell
208def _(portfolio):
209 portfolio.stats.summary()
210 return
213@app.cell
214def _():
215 return
218if __name__ == "__main__":
219 app.run()