Vibe Finance

1from AlgorithmImports import *
2
3class SMACrossRiskManaged(QCAlgorithm):
4    def Initialize(self):
5        self.SetStartDate(2010, 1, 1)
6        self.SetCash(100000)
7
8        self.SetBrokerageModel(BrokerageName.Alpaca, AccountType.Cash)
9
10
11        equity = self.AddEquity("SPY", Resolution.Daily, Market.USA)
12        self.symbol = equity.Symbol
13
14        self.SetBenchmark(self.symbol)
15
16        self.fast_period = 50
17        self.slow_period = 200
18        self.atr_period = 14
19
20        self.fast = self.SMA(self.symbol, self.fast_period, Resolution.Daily)
21        self.slow = self.SMA(self.symbol, self.slow_period, Resolution.Daily)
22        self.atr = self.ATR(self.symbol, self.atr_period, MovingAverageType.Simple, Resolution.Daily)
23
24
25        self.SetWarmUp(timedelta(days=self.slow_period + 20))
26
27        self.last_signal = 0  # -1 = below, 0 = neutral/unknown, 1 = above
28        self.stop_ticket = None
29
30        self.risk_fraction = 0.01   # 1% of portfolio per trade
31        self.max_exposure = 0.20    # 20% notional cap
32        self.stop_atr_multiplier = 2.0
33
34    def OnData(self, data: Slice):
35        if self.IsWarmingUp:
36            return
37
38        bar = data.Bars.get(self.symbol)
39        if bar is None:
40            return
41
42        if not (self.fast.IsReady and self.slow.IsReady and self.atr.IsReady):
43            return
44
45        price = bar.Close
46        if price <= 0:
47            return
48
49        fast_value = self.fast.Current.Value
50        slow_value = self.slow.Current.Value
51
52        if fast_value > slow_value:
53            current_signal = 1
54        elif fast_value < slow_value:
55            current_signal = -1
56        else:
57            current_signal = self.last_signal
58
59        bullish_cross = current_signal == 1 and self.last_signal <= 0
60        bearish_cross = current_signal == -1 and self.last_signal >= 0
61
62        invested = self.Portfolio[self.symbol].Invested
63
64        if bullish_cross and not invested:
65            self.EnterLong(price)
66
67        elif bearish_cross and invested:
68            self.ExitPosition("MA cross exit")
69
70        self.last_signal = current_signal
71
72    def EnterLong(self, price: float):
73        portfolio_value = self.Portfolio.TotalPortfolioValue
74        if portfolio_value <= 0:
75            return
76
77        atr_value = self.atr.Current.Value
78        if atr_value <= 0:
79            return
80
81        stop_price = price - self.stop_atr_multiplier * atr_value
82        if stop_price <= 0 or stop_price >= price:
83            stop_price = price * 0.95
84
85        risk_per_share = price - stop_price
86        if risk_per_share <= 0:
87            return
88
89        max_risk_capital = portfolio_value * self.risk_fraction
90        qty_by_risk = max_risk_capital / risk_per_share
91
92        max_notional = portfolio_value * self.max_exposure
93        qty_by_exposure = max_notional / price
94
95        raw_qty = min(qty_by_risk, qty_by_exposure)
96        if raw_qty <= 0:
97            return
98
99        qty = int(raw_qty * 0.99)
100        if qty <= 0:
101            return
102
103        self.Debug(f"Entering long {self.symbol} | Price: {price:.2f}, Qty: {qty}, Stop: {stop_price:.2f}")
104        ticket = self.MarketOrder(self.symbol, qty)
105
106        self.CancelStop()
107
108        self.stop_ticket = self.StopMarketOrder(self.symbol, -qty, stop_price)
109        self.Debug(f"Placed stop for {self.symbol} | Qty: {-qty}, Stop: {stop_price:.2f}")
110
111    def ExitPosition(self, reason: str):
112        if not self.Portfolio[self.symbol].Invested:
113            return
114
115        self.Debug(f"Exiting {self.symbol} due to: {reason}")
116        self.CancelStop()
117        self.Liquidate(self.symbol)
118
119    def CancelStop(self):
120        if self.stop_ticket is None:
121            return
122        status = self.stop_ticket.Status
123        if status in [OrderStatus.Submitted, OrderStatus.PartiallyFilled]:
124            self.Transactions.CancelOrder(self.stop_ticket.OrderId)
125        self.stop_ticket = None
126
127    def OnOrderEvent(self, orderEvent: OrderEvent):
128        if orderEvent is None:
129            return
130
131        if orderEvent.Status == OrderStatus.Filled:
132            self.Debug(f"Order filled: {orderEvent.Symbol} | "
133                       f"Qty: {orderEvent.FillQuantity} @ {orderEvent.FillPrice:.2f} | "
134                       f"Direction: {orderEvent.Direction}")
135
136        if self.stop_ticket is not None and orderEvent.OrderId == self.stop_ticket.OrderId:
137            if orderEvent.Status in [OrderStatus.Filled, OrderStatus.Canceled]:
138                self.stop_ticket = None
139