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

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# /// 

12 

13"""Experiment 3: Advanced CTA strategy with price filtering and oscillators. 

14 

15This module implements a more sophisticated trend-following strategy that 

16incorporates price filtering to handle outliers and oscillators with proper 

17scaling for more consistent signal generation across different assets. 

18""" 

19 

20import marimo 

21 

22__generated_with = "0.23.1" 

23app = marimo.App() 

24 

25with app.setup: 

26 import sys 

27 from pathlib import Path 

28 

29 import marimo as mo 

30 import polars as pl 

31 from jquantstats import Portfolio 

32 from tinycta.osc import osc 

33 from tinycta.util import vol_adj 

34 

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

36 

37 from preamble import date_col, load_prices 

38 

39 prices = load_prices(__file__) 

40 prices_only = prices.drop(date_col) 

41 

42 

43@app.cell(hide_code=True) 

44def _(): 

45 mo.md(r"""# CTA 3.0""") 

46 return 

47 

48 

49@app.cell(hide_code=True) 

50def _(): 

51 mo.md( 

52 r""" 

53 We use the system: 

54 $$\mathrm{CashPosition}=\frac{f(\mathrm{Price})}{\mathrm{Volatility(Returns)}}$$ 

55 

56 This is very problematic: 

57 * Prices may live on very different scales, hence trying to find a 

58 more universal function $f$ is almost impossible. The sign-function was 

59 a good choice as the results don't depend on the scale of the argument. 

60 * Price may come with all sorts of spikes/outliers/problems. 

61 """ 

62 ) 

63 return 

64 

65 

66@app.cell(hide_code=True) 

67def _(): 

68 mo.md( 

69 r""" 

70 We need a simple price filter process 

71 * We compute volatility-adjusted returns, filter them and compute prices from those returns. 

72 * Don't call it Winsorizing in Switzerland. We apply Huber functions. 

73 """ 

74 ) 

75 return 

76 

77 

78@app.cell(hide_code=True) 

79def _(): 

80 mo.md( 

81 r""" 

82 ### Oscillators 

83 * All prices are now following a standard arithmetic Brownian 

84 motion with std $1$. 

85 * What we want is the difference of two moving means (exponentially weighted) 

86 to have a constant std regardless of the two lengths. 

87 * An oscillator is the **scaled difference of two moving averages**. 

88 """ 

89 ) 

90 return 

91 

92 

93@app.function 

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

95 """Return the tanh oscillator of vol-adjusted cumulative price, divided by volatility.""" 

96 price_adj = vol_adj(price, vola=vola, clip=clip, min_samples=300).cum_sum() 

97 mu = osc(price_adj, fast=fast, slow=slow).tanh() 

98 vol = price.pct_change().ewm_std(com=slow, min_samples=300) 

99 return mu / vol 

100 

101 

102@app.cell 

103def _(): 

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

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

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

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

108 

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

110 

111 return fast, slow, vola, winsor 

112 

113 

114@app.cell 

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

116 signals = prices_only.select( 

117 (f(pl.all(), fast=fast.value, slow=slow.value, vola=vola.value, clip=winsor.value) * 1e5) 

118 .fill_nan(0.0) 

119 .fill_null(0.0) 

120 ) 

121 portfolio = Portfolio.from_cash_position(prices=prices, cash_position=signals, aum=1e8) 

122 return (portfolio,) 

123 

124 

125@app.cell 

126def _(portfolio): 

127 print(portfolio.stats.sharpe()) 

128 

129 

130@app.cell 

131def _(portfolio): 

132 fig = portfolio.plots.snapshot() 

133 fig 

134 return 

135 

136 

137if __name__ == "__main__": 

138 app.run()