Rolling Channel Breakout (Session H/L)
An end-to-end example of the persistent-state collection features: class-body list and dict declarations, for-loop iteration, subscript reads and writes, augmented assignment, and helper-method index writes.
The strategy tracks the high and low of every bar that falls into one of five global trading sessions (Asian, London, NY overlap, NY afternoon, NY close), then trades breakouts of the all-session range.
class GlobalSessionsBreakout(Strategy): """Track H/L of 5 global trading sessions and trade breakouts of the all-session range.""" trade = Symbol() sizing = FixedCapital()
# ── Per-session highs and lows — parallel lists indexed by session ── # idx 0 = asian (03:00 - 07:00) # idx 1 = london (09:00 - 12:00) # idx 2 = ny_overlap (16:30 - 18:00) # idx 3 = ny_afternoon (19:00 - 20:00) # idx 4 = ny_close (20:30 - 23:00) session_highs = [0.0]*5 session_lows = [999999.0]*5
# ── Per-session trade counters — a dict keyed by name, for readable reasons ── trade_counts = { "asian": 0, "london": 0, "ny_overlap": 0, "ny_afternoon": 0, "ny_close": 0, }
def execute(self): # Wall-clock as minutes since midnight, so each window check is one comparison. t = self.trade.timestamp.hour * 60 + self.trade.timestamp.minute
# ── Update per-session H/L through a helper ── # Helpers may mutate collection state via index writes — those propagate, because # the slot holds a reference both sides share. Rebinding a SCALAR inside a helper # does not, and is refused. self.update_session_extremes(t)
# ── Aggregate high-of-highs and low-of-lows by iterating the lists ── max_high = 0.0 min_low = 999999.0 for h in self.session_highs: if h > max_high: max_high = h for lo in self.session_lows: if lo < min_low: min_low = lo
# ── Break above the all-session high → long ── if self.trade.close > max_high and self.trade.is_flat and max_high > 0.0: # Tag the trade with whichever session we are in; each branch bumps the # dict counter through augmented assignment. if t >= 180 and t < 420: self.trade_counts["asian"] += 1 self.buy(self.trade, reason=f"Asian breakout above {max_high:.2f}") elif t >= 540 and t < 720: self.trade_counts["london"] += 1 self.buy(self.trade, reason=f"London breakout above {max_high:.2f}") elif t >= 990 and t < 1080: self.trade_counts["ny_overlap"] += 1 self.buy(self.trade, reason=f"NY overlap breakout above {max_high:.2f}") elif t >= 1140 and t < 1200: self.trade_counts["ny_afternoon"] += 1 self.buy(self.trade, reason=f"NY afternoon breakout above {max_high:.2f}") elif t >= 1230 and t < 1380: self.trade_counts["ny_close"] += 1 self.buy(self.trade, reason=f"NY close breakout above {max_high:.2f}")
# ── Break below the all-session low → exit ── if self.trade.close < min_low and self.trade.is_long: self.sell(self.trade, self.trade.position, reason=f"Break below all-session low {min_low:.2f}")
# ── Helpers ──────────────────────────────────────────────────────────────
def update_session_extremes(self, t): # asian: 03:00-07:00 → idx 0 if t >= 180 and t < 420: if self.trade.high > self.session_highs[0]: self.session_highs[0] = self.trade.high # ✅ index write — propagates if self.trade.low < self.session_lows[0]: self.session_lows[0] = self.trade.low
# london: 09:00-12:00 → idx 1 if t >= 540 and t < 720: if self.trade.high > self.session_highs[1]: self.session_highs[1] = self.trade.high if self.trade.low < self.session_lows[1]: self.session_lows[1] = self.trade.low
# ny_overlap: 16:30-18:00 → idx 2 if t >= 990 and t < 1080: if self.trade.high > self.session_highs[2]: self.session_highs[2] = self.trade.high if self.trade.low < self.session_lows[2]: self.session_lows[2] = self.trade.low
# ny_afternoon: 19:00-20:00 → idx 3 if t >= 1140 and t < 1200: if self.trade.high > self.session_highs[3]: self.session_highs[3] = self.trade.high if self.trade.low < self.session_lows[3]: self.session_lows[3] = self.trade.low
# ny_close: 20:30-23:00 → idx 4 if t >= 1230 and t < 1380: if self.trade.high > self.session_highs[4]: self.session_highs[4] = self.trade.high if self.trade.low < self.session_lows[4]: self.session_lows[4] = self.trade.low
return 0 # helpers return a number; this one's result is unusedWalkthrough
Why two shapes of collection
session_highs and session_lows are lists indexed by integer (0..4 → session). trade_counts is a dict keyed by string. Both are valid; pick by access pattern:
- List + integer index — smallest and fastest; the index-to-meaning mapping lives in a comment. Best when position IS the identity (a rolling buffer, a fixed grid).
- Dict + string key — self-documenting at every call site (
self.trade_counts["asian"]). Best when the names matter, e.g. because they end up in areason=string.
Mixing both in one strategy is fine. Use whichever reads better where you are.
The class-body declaration form
Three things to notice in the declarations above:
[0.0]*5is the[<literal>]*Nshortcut — the same list as[0.0, 0.0, 0.0, 0.0, 0.0]. The literal element pins the type as decimal;*5sets the initial length.[999999.0]*5uses a sentinel high value so that the first real low in a session beats it.0.0would not work for a minimum — every real price is above it.- The dict literal has string keys and
intvalues throughout. Mixed value types raiseRC2_STATE_MIXED; nested values raiseRC2_STATE_NESTED; non-string keys raiseRC2_STATE_DICT_KEY.
Helper-method index writes propagate
update_session_extremes writes self.session_highs[i] from inside a helper, and the change is visible to the next bar. The slot holds a reference to one list that both execute() and the helper see; an index write mutates that shared list.
A helper that tried self.session_highs = [0.0, 0.0, 0.0, 0.0, 0.0] instead would raise RC2_STATE_REBIND_COLLECTION — and a helper that rebinds a scalar raises RC2_HELPER_STATE_WRITE, because that write would not survive the return. The index-write rule is what lets helpers do useful collection work without that footgun.
Iteration to compute the aggregate range
for h in self.session_highs: iterates the list’s values; h is a fresh local that cannot collide with anything declared elsewhere. For dicts the same form iterates keys, which is Python’s own default — pair it with a lookup:
class SessionTotals(Strategy): trade = Symbol() sizing = FixedCapital() trade_counts = {"asian": 0, "london": 0, "ny_close": 0} total = 0
def execute(self): self.total = 0 for session_name in self.trade_counts: self.total = self.total + self.trade_counts[session_name]
if self.total > 10 and self.trade.is_flat: self.buy(self.trade, reason=f"{self.total} session breakouts so far")Augmented assignment on dict values
self.trade_counts["asian"] += 1 evaluates the key expression once before the read-modify-write.
What this example exercises
| Feature | Where |
|---|---|
Class-body list literal ([0.0]*5) |
session_highs, session_lows |
| Class-body dict literal | trade_counts |
| List subscript read | self.session_highs[0] |
| List subscript write | self.session_highs[0] = self.trade.high |
| Dict subscript write (augmented) | self.trade_counts["asian"] += 1 |
for x in self.<list> |
for h in self.session_highs |
| Helper-method collection mutation | update_session_extremes |
self.trade.timestamp.hour / .minute |
the t = … line at the top of execute() |
Common modifications
Reset H/L per day
The version above accumulates lifetime per-session extremes — the Asian high keeps climbing across days. For “today only”, keep a last_minute scalar and clear the slot when the clock crosses each session start:
class DailyReset(Strategy): trade = Symbol() sizing = FixedCapital() session_highs = [0.0]*5 session_lows = [999999.0]*5 last_minute = 0.0
def execute(self): t = self.trade.timestamp.hour * 60 + self.trade.timestamp.minute
# Detect the transition INTO the asian window and clear that slot. if t >= 180 and self.last_minute < 180: self.session_highs[0] = 0.0 self.session_lows[0] = 999999.0
# (…the same pattern for the other four sessions)
self.last_minute = t
if self.trade.high > self.session_highs[0]: self.session_highs[0] = self.trade.high if self.trade.close > self.session_highs[0] and self.trade.is_flat: self.buy(self.trade)Trade only the first breakout per session
Add a dict[str, bool] recording which sessions have already fired:
class FirstBreakoutOnly(Strategy): trade = Symbol() sizing = FixedCapital() session_highs = [0.0]*5 already_traded = { "asian": False, "london": False, "ny_overlap": False, "ny_afternoon": False, "ny_close": False, }
def execute(self): t = self.trade.timestamp.hour * 60 + self.trade.timestamp.minute
if self.trade.close > self.session_highs[0] and self.trade.is_flat: if t >= 180 and t < 420 and not self.already_traded["asian"]: self.already_traded["asian"] = True self.buy(self.trade, reason="First asian breakout") # (…the same pattern for the other sessions)Common diagnostics
| If you see | … the most likely cause |
|---|---|
RC2_ASSIGN_LHS |
You wrote self.list[i] = … without declaring the list in the class body — or the slot is a scalar. |
RC2_STATE_REBIND_COLLECTION |
You wrote self.list = [...] or self.dict = {...} inside execute(). Mutate by index or key instead. |
RC2_STATE_EMPTY_LIST |
You wrote name = [] in the class body. Use [0.0]*0 to pin the element type. |
RC2_HELPER_STATE_WRITE |
A helper rebound a scalar. Move that assignment to execute(); collection index and key writes from helpers are fine. |