Coverage for book/marimo/notebooks/Experiment4.py: 100%

1# /// script 

2# requires-python = ">=3.12" 

3# dependencies = [ 

4# "marimo==0.24.0", 

5# "numpy==2.4.6", 

6# "plotly==6.9.0", 

7# "polars==1.44.1", 

8# "jquantstats==0.11.0", 

9# "tinycta==0.14.0" 

10# ] 

11# 

12# [tool.ty.environment] 

13# # ``from preamble import ...`` resolves at runtime via the sys.path.insert in the 

14# # setup cell below. ty analyses a PEP 723 script in isolation from the project, so 

15# # pyproject.toml's [tool.ty.environment] never reaches this file and the path has 

16# # to be declared here. Preserve this table if marimo rewrites the header. 

17# extra-paths = ["."] 

18# /// 

19 

20"""Experiment 4: CTA strategy with optimization and risk scaling. 

21 

22This module demonstrates a more advanced trend-following strategy that 

23incorporates portfolio optimization techniques and risk scaling to 

24improve performance and risk-adjusted returns. 

25""" 

26 

27import marimo 

28 

29__generated_with = "0.23.1" 

30app = marimo.App() 

31 

32with app.setup: 

33 import sys 

34 from pathlib import Path 

35 

36 import marimo as mo 

37 import numpy as np 

38 import polars as pl 

39 from jquantstats import Portfolio 

40 from tinycta.osc import osc 

41 from tinycta.util import vol_adj 

42 

43 sys.path.insert(0, str(Path(__file__).parent)) 

44 

45 from preamble import date_col, load_prices 

46 

47 prices = load_prices(__file__) 

48 prices_only = prices.drop(date_col) 

49 assets = prices_only.columns 

50 

51 

52@app.cell(hide_code=True) 

53def _(): 

54 mo.md(r"""# CTA 4.0 - Optimization 1.0""") 

55 return 

56 

57 

58@app.function 

59def f(price: "pl.Expr", fast: int = 32, slow: int = 96, vola: int = 32, clip: float = 4.2) -> "pl.Expr": 

60 """Return the tanh oscillator of vol-adjusted cumulative price.""" 

61 return osc(vol_adj(price, vola=vola, clip=clip, min_samples=300).cum_sum(), fast=fast, slow=slow).tanh() 

62 

63 

64@app.cell 

65def _(): 

66 fast = mo.ui.slider(4, 192, step=4, value=32, label="Fast Moving Average") 

67 slow = mo.ui.slider(4, 192, step=4, value=96, label="Slow Moving Average") 

68 vola = mo.ui.slider(4, 192, step=4, value=32, label="Volatility") 

69 winsor = mo.ui.slider(1.0, 6.0, step=0.1, value=4.2, label="Winsorizing") 

70 

71 mo.vstack([fast, slow, vola, winsor]) 

72 

73 return fast, slow, vola, winsor 

74 

75 

76@app.cell 

77def _(fast, slow, vola, winsor): 

78 mu_np = prices_only.select( 

79 f(pl.all(), fast=fast.value, slow=slow.value, vola=vola.value, clip=winsor.value) 

80 ).to_numpy() 

81 volax_np = prices_only.select( 

82 pl.all().fill_nan(None).pct_change().ewm_std(com=vola.value, min_samples=vola.value) 

83 ).to_numpy() 

84 euclid_norm = np.sqrt(np.nansum(mu_np**2, axis=1, keepdims=True)) 

85 euclid_norm[euclid_norm == 0] = np.nan 

86 risk_scaled_np = mu_np / euclid_norm 

87 

88 pos_np = np.nan_to_num(5e5 * risk_scaled_np / volax_np, nan=0.0) 

89 portfolio = Portfolio.from_cash_position( 

90 prices=prices, 

91 cash_position=pl.concat( 

92 [prices.select(date_col), pl.from_numpy(pos_np, schema=dict.fromkeys(assets, pl.Float64))], 

93 how="horizontal_extend", 

94 ), 

95 aum=1e8, 

96 ) 

97 return (portfolio,) 

98 

99 

100@app.cell 

101def _(portfolio): 

102 print(portfolio.stats.sharpe()) 

103 

104 

105@app.cell 

106def _(portfolio): 

107 fig = portfolio.plots.snapshot() 

108 fig 

109 return 

110 

111 

112if __name__ == "__main__": 

113 app.run()