1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
|
"""Tests for MOC (Market on Close) strategy."""
import sys
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
from datetime import datetime, timezone
from decimal import Decimal
from shared.models import Candle, OrderSide
from strategies.moc_strategy import MocStrategy
def _candle(price, hour=20, minute=0, volume=100.0, day=1, open_price=None):
op = open_price if open_price is not None else price - 1 # Default: bullish
return Candle(
symbol="AAPL",
timeframe="5Min",
open_time=datetime(2025, 1, day, hour, minute, tzinfo=timezone.utc),
open=Decimal(str(op)),
high=Decimal(str(price + 1)),
low=Decimal(str(min(op, price) - 1)),
close=Decimal(str(price)),
volume=Decimal(str(volume)),
)
def _make_strategy(**overrides):
s = MocStrategy()
params = {
"quantity_pct": 0.2,
"stop_loss_pct": 2.0,
"rsi_min": 30,
"rsi_max": 70, # Wider for tests
"ema_period": 5,
"volume_avg_period": 5,
"min_volume_ratio": 0.5,
"buy_start_utc": 19,
"buy_end_utc": 21,
"sell_start_utc": 13,
"sell_end_utc": 15,
"max_positions": 5,
}
params.update(overrides)
s.configure(params)
return s
def test_moc_warmup_period():
s = _make_strategy(ema_period=20, volume_avg_period=15)
assert s.warmup_period == 21
def test_moc_no_signal_outside_buy_window():
s = _make_strategy()
# Hour 12 UTC — not in buy (19-21) or sell (13-15) window
for i in range(10):
sig = s.on_candle(_candle(150 + i, hour=12, minute=i * 5))
assert sig is None
def test_moc_buy_signal_in_window():
s = _make_strategy(ema_period=3)
# Build up history with some oscillation so RSI settles in the 30-70 range
prices = [150, 149, 151, 148, 152, 149, 150, 151, 148, 150,
149, 151, 150, 152, 151, 153, 152, 154, 153, 155]
signals = []
for i, p in enumerate(prices):
sig = s.on_candle(_candle(p, hour=20, minute=i * 2, volume=200.0))
if sig is not None:
signals.append(sig)
buy_signals = [sig for sig in signals if sig.side == OrderSide.BUY]
assert len(buy_signals) > 0
assert buy_signals[0].strategy == "moc"
def test_moc_sell_at_open():
s = _make_strategy(ema_period=3)
# Force entry
for i in range(10):
s.on_candle(_candle(150 + i, hour=20, minute=i * 3, volume=200.0))
if s._in_position:
# Next day, sell window
sig = s.on_candle(_candle(155, hour=14, minute=0, day=2))
assert sig is not None
assert sig.side == OrderSide.SELL
assert "MOC sell" in sig.reason
def test_moc_stop_loss():
s = _make_strategy(ema_period=3, stop_loss_pct=1.0)
for i in range(10):
s.on_candle(_candle(150 + i, hour=20, minute=i * 3, volume=200.0))
if s._in_position:
drop_price = s._entry_price * 0.98 # -2%
sig = s.on_candle(_candle(drop_price, hour=22, minute=0))
if sig is not None:
assert sig.side == OrderSide.SELL
assert "stop loss" in sig.reason
def test_moc_no_buy_on_bearish_candle():
s = _make_strategy(ema_period=3)
for i in range(8):
s.on_candle(_candle(150, hour=20, minute=i * 3, volume=200.0))
# Bearish candle (open > close)
sig = s.on_candle(_candle(149, hour=20, minute=30, open_price=151))
# May or may not signal depending on other criteria, but bearish should reduce chances
# Just verify no crash
def test_moc_only_one_buy_per_day():
s = _make_strategy(ema_period=3)
buy_count = 0
for i in range(20):
sig = s.on_candle(_candle(150 + i * 0.3, hour=20, minute=i * 2, volume=200.0))
if sig is not None and sig.side == OrderSide.BUY:
buy_count += 1
assert buy_count <= 1
def test_moc_reset():
s = _make_strategy()
s.on_candle(_candle(150, hour=20))
s._in_position = True
s.reset()
assert not s._in_position
assert len(s._closes) == 0
assert not s._bought_today
|
