For agentic workers: REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (
- [ ]) syntax for tracking.
Goal: Test, on five years of Tickstory 5m data and behind pre-registered criteria, whether "ZTD H1 bias + any 5m pullback + pivot-break stop entry" makes money.
Architecture: A new pyakao.entry5m package: bias.py drives the existing ZTD port once per closed H1 bar and keeps the active bias; setup.py finds 5m pivots and pullbacks and prices the stop order; execution.py fills and exits with R accounting; runner.py walks the 5m bars once and wires the three; report.py computes the pre-registered table and verdict; screen.py is the CLI. Two small additions to pyakao/ztd.py expose the contradiction event and the zero-touch extreme.
Tech Stack: Python 3.11+ (the machine runs 3.13), pytest, the existing pyakao package (ztd.py, zms/htf.py, zms/metrics.py, feeds/tickstory.py, feeds/base.py, pinets.py). No new dependencies.
Spec: entry5m/docs/2026-09-06-entry5m-design.md (binding; section numbers below refer to it).
pyakao/src/pyakao/entry5m/ plus the listed edits to pyakao/src/pyakao/ztd.py, pyakao/src/pyakao/feeds/base.py, pyakao/tests/test_ztd_golden.py. The Pine script ztd_divergence.pine is NOT touched.HtfSeries.is_new(i)).pivot_n=3, min_pb_bars=3, timeout_h1=96, min_rr=1.0, pending_max_bars=48, max_hold_bars=0, ZtdConfig() defaults.pivot_n in {2,3,5} x min_rr in {1.0,1.5} is reported, never judged.pyakao.feeds.base.spec_for(symbol): mintick, slippage_ticks (EURUSD/GBPUSD 2 ticks, XAUUSD 18, EURJPY 3 after Task 1). Slip = slippage_ticks * mintick, applied on stop fills and stop-loss exits and bias-end exits, not on target fills.D:/tickstory/History data/<SYMBOL>_mt5_bars.csv via load_tickstory(path, timeframe="5m"). Instruments EURUSD (primary), GBPUSD (same test as EURUSD), XAUUSD, EURJPY (non-USD control). EURGBP is unusable (240 minute bars).D:\vwap tpo pine\pyakao with python -m pytest -q. Baseline 202 tests green; every task ends green.Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>. Never stage planning/, pyakao/planning/, *.bak, .superpowers/, or data JSON/CSV."RDIV" and "HDIV"; sides are ints -1 (short) and +1 (long); reasons are the exact strings listed in Task 4/5.| File | Responsibility |
|---|---|
pyakao/src/pyakao/ztd.py (modify) |
ZtdRuleOut + ZtdRow gain mid_lo_b/mid_hi_u and contraB/contraU/midLoB/midHiU; ZtdRule tracks the zero-touch extreme |
pyakao/src/pyakao/feeds/base.py (modify) |
EURJPY registry entry |
pyakao/tests/test_ztd_golden.py (modify) |
skip the four new fields |
pyakao/src/pyakao/entry5m/__init__.py |
exports Entry5mConfig, run_entry5m, Entry5mResult |
pyakao/src/pyakao/entry5m/config.py |
Entry5mConfig dataclass |
pyakao/src/pyakao/entry5m/bias.py |
Bias, BiasTracker (H1 ZTD -> bias lifecycle) |
pyakao/src/pyakao/entry5m/setup.py |
Pullback, PendingOrder, PullbackScanner (5m pivots, pullbacks, order pricing) |
pyakao/src/pyakao/entry5m/execution.py |
Position, Trade, Broker (fills, exits, R accounting) |
pyakao/src/pyakao/entry5m/runner.py |
Entry5mResult, run_entry5m (one pass over 5m bars) |
pyakao/src/pyakao/entry5m/report.py |
summarize, judge, format_table, format_sensitivity |
pyakao/src/pyakao/entry5m/screen.py |
CLI: load Tickstory, run primary cell + sensitivity, print/write |
pyakao/tests/test_entry5m_bias.py, test_entry5m_setup.py, test_entry5m_execution.py, test_entry5m_runner.py, test_entry5m_report.py |
unit + integration tests |
entry5m/research/2026-09-06-entry5m-screen.md |
the run's output and verdict (Task 7) |
entry5m/README.md |
sub-project index (Task 7) |
Files:
pyakao/src/pyakao/ztd.py (ZtdRuleOut ~line 127-161, ZtdRule.__init__ ~line 271-305, bear block ~line 331-417, bull block ~line 421-503, ZtdRuleOut(...) construction ~line 532-558, ZtdRow ~line 84-124, ZtdRow(...) construction ~line 634-648)pyakao/src/pyakao/feeds/base.py:97-108 (REGISTRY)pyakao/tests/test_ztd_golden.py (every loop over __dataclass_fields__)pyakao/tests/test_ztd.py (append)Interfaces:
Consumes: nothing new.
Produces: ZtdRow.contraB: int, ZtdRow.contraU: int, ZtdRow.midLoB: float, ZtdRow.midHiU: float (na when not applicable); ZtdRuleOut.mid_lo_b, ZtdRuleOut.mid_hi_u; spec_for("EURJPY") -> SymbolSpec("EURJPY", 0.001, "fx", None, False, "JPY", 3).
[ ] Step 1: Write the failing rule-level tests (append to pyakao/tests/test_ztd.py, after test_not_ready_freezes_the_range)
# ---------------------------------------------------------------- rule: zero-touch extreme (entry5m)
def test_zero_touch_extreme_tracks_the_lowest_low_while_locked():
r = ZtdRule(cfg())
out = _lock_bear(r) # lock bar: low 10.0
assert out.mid_lo_b == 10.0
out = step(r, 10.5, 9.0, -0.8) # still locked, lower low
assert out.mid_lo_b == 9.0
out = step(r, 10.5, 9.5, -0.2) # higher low: unchanged
assert out.mid_lo_b == 9.0
fire = step(r, 13.0, 12.0, 0.5) # range 2 opens + RDIV- fires
assert fire.sig_bear == 2 and fire.bear_state == 3
assert fire.mid_lo_b == 9.0 # carried into range 2 for the fire bar
assert na(fire.mid_hi_u) # bull side never locked
def test_zero_touch_extreme_clears_when_the_reference_is_dropped():
r = ZtdRule(cfg(timeout_bars=5))
_lock_bear(r)
for _ in range(6):
out = step(r, 10.0, 9.0, -0.5)
assert out.bear_state == 0 and na(out.mid_lo_b)
def test_zero_touch_extreme_restarts_on_a_new_lock():
r = ZtdRule(cfg())
_lock_bear(r) # mid 10.0
step(r, 11.0, 10.0, 0.5) # unqualified range 2, no fire
out = step(r, 10.0, 8.0, -0.5) # zero touch: locked again (Until superseded)
assert out.bear_state == 2 and out.mid_lo_b == 8.0
def test_bull_zero_touch_extreme_mirrors():
r = ZtdRule(cfg())
step(r, 11.0, 10.0, -1.0, arm_u=True)
step(r, 9.0, 8.0, -2.5)
out = step(r, 10.0, 9.0, 0.5) # lock: high 10.0
assert out.bull_state == 2 and out.mid_hi_u == 10.0
out = step(r, 12.0, 9.5, 0.8)
assert out.mid_hi_u == 12.0
fire = step(r, 8.0, 7.0, -0.5) # lower low, higher z -> RDIV+
assert fire.sig_bull == -2 and fire.mid_hi_u == 12.0
And the engine plumbing test (append after test_sig_prefers_the_bear_side):
def test_row_exposes_contradiction_events_and_zero_touch_extremes():
bars = _wave_bars(600)
eng = ZtdEngine(_fast_cfg())
captured = []
orig = eng.rule.update
def spy(*a, **k):
out = orig(*a, **k)
captured.append(out)
return out
eng.rule.update = spy
rows = []
for i in range(len(bars)):
rows.append(eng.update(bars.highs[i], bars.lows[i], bars.closes[i], bars.volumes[i],
int(bars.times[i].timestamp())))
for row, out in zip(rows, captured):
assert row.contraB == (1 if out.bear_contra_now else 0)
assert row.contraU == (1 if out.bull_contra_now else 0)
assert (na(row.midLoB) and na(out.mid_lo_b)) or row.midLoB == out.mid_lo_b
assert (na(row.midHiU) and na(out.mid_hi_u)) or row.midHiU == out.mid_hi_u
assert all(not na(r.midLoB) for r in rows if r.sigBear)
assert all(not na(r.midHiU) for r in rows if r.sigBull)
def test_eurjpy_spec_is_registered():
from pyakao.feeds.base import spec_for
s = spec_for("EURJPY")
assert (s.mintick, s.slippage_ticks, s.currency) == (0.001, 3, "JPY")
Run: cd "D:\vwap tpo pine\pyakao" && python -m pytest tests/test_ztd.py -q -k "zero_touch or contradiction_events or eurjpy"
Expected: FAIL (AttributeError: 'ZtdRuleOut' object has no attribute 'mid_lo_b', and the EURJPY slippage is 2 from the fallback).
ztd.py(a) ZtdRuleOut: after bull_attn_now: bool add
mid_lo_b: float # zero-touch extreme (lowest low while the bear side is locked)
mid_hi_u: float # mirror: highest high while the bull side is locked
(b) ZtdRule.__init__: next to self.bear_drawn_tr = NA add self.mid_lo_b = NA; next to self.bull_drawn_tr = NA add self.mid_hi_u = NA.
(c) Bear block, immediately after bear_locked_now = self.bear_state == 2 and prev_bear_state != 2:
# zero-touch extreme (entry5m target): lowest low while locked (state 2),
# carried through range 2 so the fire bar can read it, cleared when the
# side leaves for state 0/1. A fresh lock restarts it.
if self.bear_state == 2:
self.mid_lo_b = low if prev_bear_state != 2 else min(self.mid_lo_b, low)
elif self.bear_state != 3:
self.mid_lo_b = NA
(d) Bull block, immediately after bull_locked_now = ...:
if self.bull_state == 2:
self.mid_hi_u = high if prev_bull_state != 2 else max(self.mid_hi_u, high)
elif self.bull_state != 3:
self.mid_hi_u = NA
(e) return ZtdRuleOut(...): add mid_lo_b=self.mid_lo_b, mid_hi_u=self.mid_hi_u,.
(f) ZtdRow: after sig: int add
# entry5m additions - not Pine plots, ignored by the golden comparison
contraB: int
contraU: int
midLoB: float
midHiU: float
(g) return ZtdRow(...) in ZtdEngine.update: add
contraB=1 if r.bear_contra_now else 0,
contraU=1 if r.bull_contra_now else 0,
midLoB=r.mid_lo_b,
midHiU=r.mid_hi_u,
(h) feeds/base.py REGISTRY: after the USDJPY line add
"EURJPY": SymbolSpec("EURJPY", 0.001, "fx", None, False, "JPY", 3),
In pyakao/tests/test_ztd_golden.py add near the other constants:
# entry5m additions on ZtdRow that are not Pine plots and have no exported column
EXTRA_FIELDS = {"contraB", "contraU", "midLoB", "midHiU"}
Change fields = [f for f in rows[0].__dataclass_fields__ if f != "time"] to ... if f != "time" and f not in EXTRA_FIELDS] (keep assert len(fields) == 37). In every other loop over rows[0].__dataclass_fields__ (there is one in test_float_agreement_is_far_inside_the_tolerances; grep for __dataclass_fields__ in tests/ and validation/), skip f in EXTRA_FIELDS the same way.
Run: cd "D:\vwap tpo pine\pyakao" && python -m pytest -q
Expected: 202 + 6 = 208 passed (the golden test still runs on the exported data in validation/data_ztd/; if that directory is absent on this machine the golden tests skip, note it in the report).
cd "D:\vwap tpo pine" && git add pyakao/src/pyakao/ztd.py pyakao/src/pyakao/feeds/base.py pyakao/tests/test_ztd.py pyakao/tests/test_ztd_golden.py
git commit -m "feat(ztd-port): expose contradiction events and the zero-touch extreme; EURJPY spec" -m "Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>"
Files:
pyakao/src/pyakao/entry5m/__init__.py, config.py, bias.pypyakao/tests/test_entry5m_bias.pyInterfaces:
Consumes: pyakao.ztd.ZtdConfig, ZtdEngine (Task 1 fields), pyakao.zms.htf.HtfBar, pyakao.pinets.na.
Produces:
Entry5mConfig (dataclass): pivot_n=3, min_pb_bars=3, timeout_h1=96, min_rr=1.0, pending_max_bars=48, max_hold_bars=0, mintick=0.00001, slippage_ticks=2, htf="1h", htf_offset_s=0, ztd: ZtdConfig, property slip -> float.Bias (dataclass): id: int, side: int, kind: str, start_h1: int, ref: float, ext: float, target: float, deadline_h1: int, start_5m: int = -1, end_5m: int = -1, end_reason: str = "".BiasTracker(cfg, engine=None): .active: Bias | None, .history: list[Bias], .no_target: int, .h1_index: int; .on_h1_close(h: HtfBar) -> list[tuple[str, Bias]] with event kinds "end" then "start"; .end_active(reason: str, at_5m: int) -> Bias | None.RecordingEngine(inner) (.rows) and ReplayEngine(rows): engine stand-ins with the same .update(...) shape, used by the screen to run the sensitivity grid without recomputing ZTD."target", "contradiction", "timeout", "replaced".[ ] Step 1: Write the failing tests (pyakao/tests/test_entry5m_bias.py)
"""BiasTracker: H1 ZTD rows -> bias lifecycle (spec section 4).
The ZTD engine is replaced by a scripted stand-in so each scenario is a
handful of rows instead of a 2000-bar warm-up.
"""
from datetime import datetime, timedelta, timezone
from types import SimpleNamespace
from pyakao.entry5m.bias import Bias, BiasTracker
from pyakao.entry5m.config import Entry5mConfig
from pyakao.zms.htf import HtfBar
NAN = float("nan")
T0 = datetime(2024, 1, 1, tzinfo=timezone.utc)
def row(**kw):
base = dict(sigBear=0, sigBull=0, contraB=0, contraU=0, ext=0, curHi=NAN, curLo=NAN,
fRefB=NAN, fRefU=NAN, midLoB=NAN, midHiU=NAN)
base.update(kw)
return SimpleNamespace(**base)
class Scripted:
"""engine.update(...) returns the scripted row for that call, a blank row after."""
def __init__(self, rows):
self.rows = rows
self.calls = []
def update(self, high, low, close, volume, time=0):
self.calls.append((high, low, close, volume, time))
k = len(self.calls) - 1
return self.rows[k] if k < len(self.rows) else row()
def h1(k, hi=101.0, lo=99.0):
return HtfBar(T0 + timedelta(hours=k), 100.0, hi, lo, 100.0, 1.0)
def drive(rows, cfg=None):
tr = BiasTracker(cfg or Entry5mConfig(), engine=Scripted(rows))
events = []
for k in range(len(rows)):
events.append(tr.on_h1_close(h1(k)))
return tr, events
def test_bear_fire_starts_a_short_bias_with_the_zero_touch_target():
tr, ev = drive([row(), row(sigBear=2, fRefB=105.0, curHi=106.0, midLoB=97.0)])
assert ev[0] == []
(kind, b), = ev[1]
assert kind == "start" and tr.active is b
assert (b.side, b.kind, b.start_h1, b.ref, b.ext, b.target) == (-1, "RDIV", 1, 105.0, 106.0, 97.0)
assert b.deadline_h1 == 1 + 96 and b.id == 1
def test_hidden_bull_fire_starts_a_long_bias():
tr, ev = drive([row(sigBull=-1, fRefU=95.0, curLo=94.0, midHiU=103.0)])
(kind, b), = ev[0]
assert (b.side, b.kind, b.target, b.ext) == (1, "HDIV", 103.0, 94.0)
def test_fire_without_a_target_is_ignored_and_counted():
tr, ev = drive([row(sigBear=2, fRefB=105.0, curHi=106.0)])
assert ev[0] == [] and tr.active is None and tr.no_target == 1
def test_engine_is_fed_the_h1_bar_values_and_open_time():
eng = Scripted([row()])
tr = BiasTracker(Entry5mConfig(), engine=eng)
tr.on_h1_close(h1(3, hi=110.0, lo=90.0))
assert eng.calls == [(110.0, 90.0, 100.0, 1.0, int((T0 + timedelta(hours=3)).timestamp()))]
assert tr.h1_index == 0
def test_contradiction_ends_the_bias():
tr, ev = drive([row(sigBear=2, fRefB=105.0, curHi=106.0, midLoB=97.0), row(), row(contraB=1)])
(kind, b), = ev[2]
assert kind == "end" and b.end_reason == "contradiction" and tr.active is None
assert tr.history == [b]
def test_opposite_side_contradiction_does_not_end_it():
tr, ev = drive([row(sigBear=2, fRefB=105.0, curHi=106.0, midLoB=97.0), row(contraU=1)])
assert ev[1] == [] and tr.active is not None
def test_timeout_counts_h1_bars_from_the_fire():
cfg = Entry5mConfig(timeout_h1=2)
tr, ev = drive([row(sigBear=2, fRefB=105.0, curHi=106.0, midLoB=97.0), row(), row(), row()], cfg)
assert ev[1] == [] and ev[2] == [] # k=1: 1 <= 2, k=2: 2 <= 2
(kind, b), = ev[3] # k=3 > deadline 2
assert kind == "end" and b.end_reason == "timeout"
def test_extension_renews_the_deadline_and_moves_ext_but_not_the_target():
cfg = Entry5mConfig(timeout_h1=2)
tr, ev = drive([row(sigBear=2, fRefB=105.0, curHi=106.0, midLoB=97.0),
row(ext=1, curHi=107.5), row(), row(), row()], cfg)
b = tr.history[0] if tr.history else tr.active
assert ev[1] == [] and b.ext == 107.5 and b.target == 97.0 and b.deadline_h1 == 1 + 2
assert ev[2] == [] and ev[3] == []
assert ev[4] == [("end", b)] and b.end_reason == "timeout"
def test_bull_extension_is_ignored_by_a_short_bias():
cfg = Entry5mConfig(timeout_h1=2)
tr, ev = drive([row(sigBear=2, fRefB=105.0, curHi=106.0, midLoB=97.0), row(ext=-1, curLo=90.0), row(), row()], cfg)
assert tr.active is None and tr.history[0].end_reason == "timeout" and tr.history[0].ext == 106.0
def test_opposite_fire_replaces_the_bias():
tr, ev = drive([row(sigBear=2, fRefB=105.0, curHi=106.0, midLoB=97.0),
row(sigBull=-2, fRefU=95.0, curLo=94.0, midHiU=103.0)])
assert [k for k, _ in ev[1]] == ["end", "start"]
old, new = ev[1][0][1], ev[1][1][1]
assert old.end_reason == "replaced" and new.side == 1 and tr.active is new and new.id == 2
def test_same_side_fire_replaces_the_record():
tr, ev = drive([row(sigBear=2, fRefB=105.0, curHi=106.0, midLoB=97.0),
row(sigBear=1, fRefB=108.0, curHi=107.0, midLoB=99.0)])
assert [k for k, _ in ev[1]] == ["end", "start"]
assert tr.active.kind == "HDIV" and tr.active.target == 99.0 and tr.history[0].end_reason == "replaced"
def test_end_active_from_the_5m_clock():
tr, _ = drive([row(sigBear=2, fRefB=105.0, curHi=106.0, midLoB=97.0)])
b = tr.end_active("target", at_5m=123)
assert b.end_reason == "target" and b.end_5m == 123 and tr.active is None and tr.history == [b]
assert tr.end_active("target", at_5m=124) is None
def test_default_engine_is_a_real_ztd_engine():
from pyakao.ztd import ZtdEngine
assert isinstance(BiasTracker(Entry5mConfig()).engine, ZtdEngine)
def test_recording_and_replay_engines_reproduce_the_bias_stream():
from pyakao.entry5m.bias import RecordingEngine, ReplayEngine
rows = [row(), row(sigBear=2, fRefB=105.0, curHi=106.0, midLoB=97.0), row(), row(contraB=1)]
rec = RecordingEngine(Scripted(rows))
a = BiasTracker(Entry5mConfig(), engine=rec)
for k in range(4):
a.on_h1_close(h1(k))
assert len(rec.rows) == 4
b = BiasTracker(Entry5mConfig(), engine=ReplayEngine(rec.rows))
for k in range(4):
b.on_h1_close(h1(k))
assert [(x.side, x.target, x.end_reason) for x in a.history] == [(-1, 97.0, "contradiction")]
assert [(x.side, x.target, x.end_reason) for x in b.history] == [(-1, 97.0, "contradiction")]
Run: cd "D:\vwap tpo pine\pyakao" && python -m pytest tests/test_entry5m_bias.py -q
Expected: FAIL with ModuleNotFoundError: No module named 'pyakao.entry5m'.
config.py"""Entry5m configuration. The primary cell (spec section 8) is the defaults."""
from __future__ import annotations
from dataclasses import dataclass, field
from ..ztd import ZtdConfig
@dataclass
class Entry5mConfig:
pivot_n: int = 3 # 5m pivot strength (bars each side)
min_pb_bars: int = 3 # pullback pivot must be this many bars after its origin
timeout_h1: int = 96 # bias life in H1 bars, renewed by extensions
min_rr: float = 1.0 # reward-to-risk floor at order placement
pending_max_bars: int = 48 # unfilled stop order expires after this many 5m bars
max_hold_bars: int = 0 # 0 = no time stop
mintick: float = 0.00001
slippage_ticks: int = 2
htf: str = "1h" # the bias timeframe
htf_offset_s: int = 0 # HTF grid offset (see zms.htf.build_htf)
ztd: ZtdConfig = field(default_factory=ZtdConfig)
@property
def slip(self) -> float:
return self.slippage_ticks * self.mintick
@property
def tick(self) -> float:
return self.mintick
bias.py"""Bias layer: closed H1 bars -> ZTD -> the active bias (spec section 4).
The tracker is fed one closed H1 bar at a time and returns the events that
bar produced, in order: an "end" (contradiction / timeout / replaced) before
any "start". Target hits are decided on the 5m clock by the runner, which
calls `end_active("target", i)`.
"""
from __future__ import annotations
from dataclasses import dataclass
from ..pinets import na
from ..zms.htf import HtfBar
from ..ztd import ZtdEngine
from .config import Entry5mConfig
END_TARGET = "target"
END_CONTRADICTION = "contradiction"
END_TIMEOUT = "timeout"
END_REPLACED = "replaced"
@dataclass
class Bias:
id: int
side: int # -1 short (sigBear), +1 long (sigBull)
kind: str # "RDIV" | "HDIV"
start_h1: int # index of the H1 bar that fired
ref: float # range-1 extreme (informational)
ext: float # range-2 extreme, follows extensions
target: float # zero-touch extreme between range 1 and range 2
deadline_h1: int # last H1 index the bias may still be active on
start_5m: int = -1
end_5m: int = -1
end_reason: str = ""
class BiasTracker:
def __init__(self, cfg: Entry5mConfig, engine=None):
self.cfg = cfg
self.engine = engine if engine is not None else ZtdEngine(cfg.ztd)
self.active: Bias | None = None
self.history: list[Bias] = []
self.no_target = 0
self.h1_index = -1
self._next_id = 1
# ------------------------------------------------------------ events ---
def end_active(self, reason: str, at_5m: int) -> Bias | None:
b = self.active
if b is None:
return None
b.end_reason = reason
b.end_5m = at_5m
self.history.append(b)
self.active = None
return b
def on_h1_close(self, h: HtfBar) -> list[tuple[str, Bias]]:
self.h1_index += 1
k = self.h1_index
row = self.engine.update(h.high, h.low, h.close, h.volume, int(h.open_time.timestamp()))
events: list[tuple[str, Bias]] = []
b = self.active
if b is not None:
contra = row.contraB if b.side < 0 else row.contraU
if contra:
events.append(("end", self._end_h1(END_CONTRADICTION)))
else:
if (b.side < 0 and row.ext == 1) or (b.side > 0 and row.ext == -1):
b.ext = row.curHi if b.side < 0 else row.curLo
b.deadline_h1 = k + self.cfg.timeout_h1
if k > b.deadline_h1:
events.append(("end", self._end_h1(END_TIMEOUT)))
sig = row.sigBear if row.sigBear != 0 else row.sigBull
if sig != 0:
side = -1 if row.sigBear != 0 else 1
target = row.midLoB if side < 0 else row.midHiU
if na(target):
self.no_target += 1
else:
if self.active is not None:
events.append(("end", self._end_h1(END_REPLACED)))
nb = Bias(
id=self._next_id, side=side,
kind="RDIV" if abs(sig) == 2 else "HDIV",
start_h1=k,
ref=row.fRefB if side < 0 else row.fRefU,
ext=row.curHi if side < 0 else row.curLo,
target=target,
deadline_h1=k + self.cfg.timeout_h1,
)
self._next_id += 1
self.active = nb
events.append(("start", nb))
return events
def _end_h1(self, reason: str) -> Bias:
b = self.active
b.end_reason = reason
self.history.append(b)
self.active = None
return b
class RecordingEngine:
"""Wraps a ZTD engine and keeps every row, so the same H1 stream can be
replayed under other entry parameters without recomputing ZTD."""
def __init__(self, inner):
self.inner = inner
self.rows: list = []
def update(self, *a, **k):
row = self.inner.update(*a, **k)
self.rows.append(row)
return row
class ReplayEngine:
"""Hands back recorded rows in order. Only valid for the identical H1 bar stream."""
def __init__(self, rows: list):
self.rows = rows
self.n = 0
def update(self, *a, **k):
self.n += 1
return self.rows[self.n - 1]
Note ext on the fire bar: the spec says the range-2 extreme is fCurB/fCurU; on the fire bar curHi == fCurB (the port snapshots f_cur_b = self.cur_hi), so reading curHi gives the same number and keeps one code path for fire and extension. The end-5m index for H1-driven ends is filled in by the runner (Task 5), which knows the 5m bar.
__init__.py"""entry5m - ZTD H1 bias + 5m pullback structure-break entries (pre-registered study).
Spec: entry5m/docs/2026-09-06-entry5m-design.md
"""
from .config import Entry5mConfig
__all__ = ["Entry5mConfig"]
(run_entry5m / Entry5mResult are added to this file in Task 5.)
Run: cd "D:\vwap tpo pine\pyakao" && python -m pytest tests/test_entry5m_bias.py -q
Expected: 14 passed.
cd "D:\vwap tpo pine" && git add pyakao/src/pyakao/entry5m/__init__.py pyakao/src/pyakao/entry5m/config.py pyakao/src/pyakao/entry5m/bias.py pyakao/tests/test_entry5m_bias.py
git commit -m "feat(entry5m): config and H1 bias tracker on the ZTD port" -m "Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>"
Files:
pyakao/src/pyakao/entry5m/setup.pypyakao/tests/test_entry5m_setup.pyInterfaces:
Consumes: pyakao.pinets.PivotHigh, PivotLow, na; Entry5mConfig (Task 2); Bias (Task 2, only .side, .target, .id).
Produces:
Pullback (dataclass): side: int, pivot_bar: int, pivot_extreme: float, origin_bar: int, origin_extreme: float, confirm_bar: int.PendingOrder (dataclass): side: int, entry: float, stop: float, target: float, placed_bar: int, pb_bar: int, pb_extreme: float, origin_bar: int, bias_id: int, kind: str.PullbackScanner(cfg): .set_bias(side: int, start_bar: int), .clear_bias(), .on_bar(i: int, high: float, low: float) -> tuple[Pullback | None, str] where the string is "", "too_short" or "ok"; .order_for(pb: Pullback, bias: Bias) -> tuple[PendingOrder | None, str] with reasons "ok", "rr_rejected", "already_broken".[ ] Step 1: Write the failing tests (pyakao/tests/test_entry5m_setup.py)
"""PullbackScanner: 5m pivots -> pullbacks -> stop orders (spec section 5)."""
import pytest
from pyakao.entry5m.bias import Bias
from pyakao.entry5m.config import Entry5mConfig
from pyakao.entry5m.setup import PendingOrder, Pullback, PullbackScanner
def cfg(**kw):
base = dict(mintick=0.01, slippage_ticks=0)
base.update(kw)
return Entry5mConfig(**base)
def short_bias(target=90.0):
return Bias(id=7, side=-1, kind="RDIV", start_h1=0, ref=105.0, ext=106.0, target=target, deadline_h1=96)
def long_bias(target=110.0):
return Bias(id=8, side=1, kind="HDIV", start_h1=0, ref=95.0, ext=94.0, target=target, deadline_h1=96)
# a short-bias scenario: bias starts at bar 0; decline to a low of 95 at bar 4,
# pullback to a high of 98 at bar 8 (confirmed at bar 11)
SHORT_HL = [
(100.2, 99.0), (99.4, 98.0), (98.4, 97.0), (97.4, 96.0), (96.4, 95.0), # 0..4
(96.5, 95.2), (97.2, 96.0), (97.6, 96.8), (98.0, 97.2), # 5..8
(97.9, 97.0), (97.5, 96.6), (97.0, 96.2), # 9..11
]
def feed(sc, hl, start=0):
out = []
for i, (h, l) in enumerate(hl, start):
out.append(sc.on_bar(i, h, l))
return out
def test_pivot_high_confirms_n_bars_later_and_yields_a_pullback():
sc = PullbackScanner(cfg())
sc.set_bias(-1, 0)
res = feed(sc, SHORT_HL)
assert all(pb is None for pb, _ in res[:11])
pb, why = res[11]
assert why == "ok"
assert pb == Pullback(side=-1, pivot_bar=8, pivot_extreme=98.0, origin_bar=4, origin_extreme=95.0, confirm_bar=11)
def test_no_bias_means_no_pullbacks():
sc = PullbackScanner(cfg())
res = feed(sc, SHORT_HL)
assert all(pb is None and why == "" for pb, why in res)
def test_long_bias_uses_pivot_lows_not_highs():
sc = PullbackScanner(cfg())
sc.set_bias(1, 0)
res = feed(sc, SHORT_HL)
pb, why = res[7] # pivot low 95.0 at bar 4, confirmed at 7
assert why == "ok" and (pb.side, pb.pivot_bar, pb.origin_bar, pb.origin_extreme) == (1, 4, 0, 100.2)
assert res[11] == (None, "") # the pivot HIGH at bar 8 is not a long pullback
def test_pullback_must_be_old_enough():
sc = PullbackScanner(cfg(min_pb_bars=5))
sc.set_bias(-1, 0)
pb, why = feed(sc, SHORT_HL)[11]
assert pb is None and why == "too_short" # 8 - 4 = 4 < 5
def test_pivot_before_the_bias_start_does_not_count():
sc = PullbackScanner(cfg())
sc.set_bias(-1, 9) # bias starts after the pivot bar 8
pb, why = feed(sc, SHORT_HL)[11]
assert pb is None and why == ""
def test_origin_search_starts_after_the_previous_pivot_high():
# bars 0..11 as SHORT_HL, then a second pullback: lows never revisit 95
extra = [(96.8, 95.5), (96.2, 95.8), (96.9, 96.0), (97.3, 96.4), (97.0, 96.2), (96.6, 95.9), (96.4, 95.7)] # 12..18
sc = PullbackScanner(cfg())
sc.set_bias(-1, 0)
res = feed(sc, SHORT_HL + extra)
pb, why = res[18] # pivot high 97.3 at bar 15, confirmed 18
assert why == "ok" and pb.pivot_bar == 15
assert pb.origin_bar == 12 and pb.origin_extreme == 95.5 # min over bars 9..15, not bar 4
def test_order_prices_for_a_short_pullback():
sc = PullbackScanner(cfg())
sc.set_bias(-1, 0)
pb, _ = feed(sc, SHORT_HL)[11]
order, why = sc.order_for(pb, short_bias(target=90.0))
assert why == "ok"
assert order == PendingOrder(side=-1, entry=pytest.approx(94.99), stop=pytest.approx(98.01), target=90.0,
placed_bar=11, pb_bar=8, pb_extreme=98.0, origin_bar=4, bias_id=7, kind="RDIV")
def test_rr_floor_rejects_a_close_target():
sc = PullbackScanner(cfg(min_rr=1.0))
sc.set_bias(-1, 0)
pb, _ = feed(sc, SHORT_HL)[11]
order, why = sc.order_for(pb, short_bias(target=93.0)) # (94.99-93)/(98.01-94.99) = 0.66
assert order is None and why == "rr_rejected"
def test_level_already_broken_between_pivot_and_confirmation():
hl = SHORT_HL[:9] + [(97.9, 94.5), (97.5, 96.6), (97.0, 96.2)] # bar 9 low 94.5 <= 94.99
sc = PullbackScanner(cfg())
sc.set_bias(-1, 0)
pb, why = feed(sc, hl)[11]
assert why == "ok"
order, why = sc.order_for(pb, short_bias())
assert order is None and why == "already_broken"
def test_long_side_mirrors():
hl = [(200.0 - l, 200.0 - h) for (h, l) in SHORT_HL]
# mirror around 100: pivot low 102 at bar 8, origin high 105 at bar 4
sc = PullbackScanner(cfg())
sc.set_bias(1, 0)
res = feed(sc, hl)
pb, why = res[11]
assert why == "ok"
assert (pb.side, pb.pivot_bar, pb.pivot_extreme, pb.origin_bar, pb.origin_extreme) == (1, 8, 102.0, 4, 105.0)
order, why = sc.order_for(pb, long_bias(target=110.0))
assert why == "ok" and order.entry == pytest.approx(105.01) and order.stop == pytest.approx(101.99)
def test_clear_bias_resets_the_previous_pivot_boundary():
sc = PullbackScanner(cfg())
sc.set_bias(-1, 0)
feed(sc, SHORT_HL)
sc.clear_bias()
sc.set_bias(-1, 12)
assert sc.on_bar(12, 96.0, 95.0) == (None, "")
Run: cd "D:\vwap tpo pine\pyakao" && python -m pytest tests/test_entry5m_setup.py -q
Expected: FAIL with ModuleNotFoundError: No module named 'pyakao.entry5m.setup'.
setup.py"""Setup layer: 5m pivots, pullbacks and the stop order they price (spec section 5).
Rules are written for the short side; the long side mirrors every extreme
and inequality. Nothing here looks past the current bar: a pivot at bar p is
only known at bar p + n, and the order is priced at that confirmation bar.
"""
from __future__ import annotations
from dataclasses import dataclass
from ..pinets import PivotHigh, PivotLow, na
from .bias import Bias
from .config import Entry5mConfig
@dataclass(frozen=True)
class Pullback:
side: int
pivot_bar: int # the pullback's extreme (pivot high for a short bias)
pivot_extreme: float
origin_bar: int # the swing the pullback formed (lowest low before it)
origin_extreme: float
confirm_bar: int # pivot_bar + n
@dataclass(frozen=True)
class PendingOrder:
side: int
entry: float # stop-entry price (one tick beyond the origin)
stop: float # stop-loss (one tick beyond the pullback extreme)
target: float
placed_bar: int
pb_bar: int
pb_extreme: float
origin_bar: int
bias_id: int
kind: str
class PullbackScanner:
def __init__(self, cfg: Entry5mConfig):
self.cfg = cfg
n = cfg.pivot_n
self._ph = PivotHigh(n, n)
self._pl = PivotLow(n, n)
self.highs: list[float] = []
self.lows: list[float] = []
self.side = 0
self.start_bar = -1
self._last_pivot = -1 # previous confirmed same-kind pivot bar since the bias started
# ------------------------------------------------------------- bias ---
def set_bias(self, side: int, start_bar: int) -> None:
self.side = side
self.start_bar = start_bar
self._last_pivot = -1
def clear_bias(self) -> None:
self.side = 0
self.start_bar = -1
self._last_pivot = -1
# ------------------------------------------------------------- bars ---
def on_bar(self, i: int, high: float, low: float) -> tuple[Pullback | None, str]:
self.highs.append(high)
self.lows.append(low)
assert len(self.highs) == i + 1, "bars must be fed contiguously from 0"
ph = self._ph.update(high)
pl = self._pl.update(low)
n = self.cfg.pivot_n
if self.side == 0:
return None, ""
piv = ph if self.side < 0 else pl
if na(piv):
return None, ""
p = i - n
if p < self.start_bar:
return None, ""
q = self._last_pivot + 1 if self._last_pivot >= 0 else self.start_bar
self._last_pivot = p
if self.side < 0:
origin = min(range(q, p + 1), key=lambda j: (self.lows[j], j))
origin_px = self.lows[origin]
else:
origin = min(range(q, p + 1), key=lambda j: (-self.highs[j], j))
origin_px = self.highs[origin]
if p - origin < self.cfg.min_pb_bars:
return None, "too_short"
return Pullback(self.side, p, piv, origin, origin_px, i), "ok"
# ------------------------------------------------------------ orders ---
def order_for(self, pb: Pullback, bias: Bias) -> tuple[PendingOrder | None, str]:
tick = self.cfg.tick
c = pb.confirm_bar
if pb.side < 0:
entry = pb.origin_extreme - tick
stop = pb.pivot_extreme + tick
reward, risk = entry - bias.target, stop - entry
broken = min(self.lows[pb.pivot_bar + 1: c + 1]) <= entry
else:
entry = pb.origin_extreme + tick
stop = pb.pivot_extreme - tick
reward, risk = bias.target - entry, entry - stop
broken = max(self.highs[pb.pivot_bar + 1: c + 1]) >= entry
if risk <= 0 or reward / risk < self.cfg.min_rr:
return None, "rr_rejected"
if broken:
return None, "already_broken"
return PendingOrder(pb.side, entry, stop, bias.target, c, pb.pivot_bar, pb.pivot_extreme,
pb.origin_bar, bias.id, bias.kind), "ok"
Run: cd "D:\vwap tpo pine\pyakao" && python -m pytest tests/test_entry5m_setup.py -q
Expected: 11 passed.
cd "D:\vwap tpo pine" && git add pyakao/src/pyakao/entry5m/setup.py pyakao/tests/test_entry5m_setup.py
git commit -m "feat(entry5m): 5m pivots, pullbacks and stop-order pricing" -m "Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>"
Files:
pyakao/src/pyakao/entry5m/execution.pypyakao/tests/test_entry5m_execution.pyInterfaces:
Consumes: PendingOrder (Task 3), Entry5mConfig (Task 2).
Produces:
Position (dataclass): order: PendingOrder, fill: float, fill_bar: int, fill_time: datetime, mfe: float, mae: float (price units).Trade (dataclass): side, kind, bias_id, entry_bar, exit_bar, entry_time, exit_time, entry, exit, stop, target, risk, r, reason, mfe_r, mae_r; properties bars_held, price_pnl (== r), direction (== side), comment (== kind) so pyakao.zms.metrics.build_report(trades, unit=1.0) works unchanged.Broker(cfg): .pending: PendingOrder | None, .position: Position | None, .trades: list[Trade]; .place(order) -> bool (True when it replaced a pending order); .cancel(reason: str) -> bool; .on_bar(i, t, o, h, l, c) -> list[str] returning events from "filled", "exit:stop", "exit:target", "exit:time", "cancel:extreme_taken", "cancel:expired"; .close_at(i, t, price, reason) for bias-end exits (reason "bias_end").Trade.reason: "stop", "target", "time", "bias_end".[ ] Step 1: Write the failing tests (pyakao/tests/test_entry5m_execution.py)
"""Broker: stop-entry fills, exits, R accounting (spec section 6)."""
from datetime import datetime, timedelta, timezone
import pytest
from pyakao.entry5m.config import Entry5mConfig
from pyakao.entry5m.execution import Broker, Trade
from pyakao.entry5m.setup import PendingOrder
T0 = datetime(2024, 1, 1, tzinfo=timezone.utc)
def t(i):
return T0 + timedelta(minutes=5 * i)
def cfg(**kw):
base = dict(mintick=0.01, slippage_ticks=0)
base.update(kw)
return Entry5mConfig(**base)
def short_order(entry=94.99, stop=98.01, target=90.0, placed=11):
return PendingOrder(-1, entry, stop, target, placed, 8, 98.0, 4, 7, "RDIV")
def long_order(entry=105.01, stop=101.99, target=110.0, placed=11):
return PendingOrder(1, entry, stop, target, placed, 8, 102.0, 4, 8, "HDIV")
def test_sell_stop_fills_at_the_entry_when_the_open_is_above_it():
b = Broker(cfg())
b.place(short_order())
assert b.on_bar(12, t(12), 96.4, 96.6, 94.0, 94.5) == ["filled"]
assert b.position.fill == pytest.approx(94.99) and b.position.fill_bar == 12 and b.pending is None
def test_gap_below_the_entry_fills_at_the_open():
b = Broker(cfg())
b.place(short_order())
b.on_bar(12, t(12), 94.0, 94.2, 93.0, 93.5)
assert b.position.fill == pytest.approx(94.0)
def test_slippage_worsens_stop_fills_and_stop_exits_but_not_targets():
b = Broker(cfg(slippage_ticks=2)) # slip 0.02
b.place(short_order())
b.on_bar(12, t(12), 96.4, 96.6, 94.0, 94.5)
assert b.position.fill == pytest.approx(94.97)
b.on_bar(13, t(13), 94.5, 98.5, 94.0, 98.0)
tr = b.trades[0]
assert tr.reason == "stop" and tr.exit == pytest.approx(98.03)
b2 = Broker(cfg(slippage_ticks=2))
b2.place(short_order())
b2.on_bar(12, t(12), 96.4, 96.6, 94.0, 94.5)
b2.on_bar(13, t(13), 94.5, 94.8, 89.5, 90.5)
assert b2.trades[0].reason == "target" and b2.trades[0].exit == pytest.approx(90.0)
def test_no_fill_before_the_entry_is_touched():
b = Broker(cfg())
b.place(short_order())
assert b.on_bar(12, t(12), 96.4, 96.6, 95.0, 95.5) == []
assert b.position is None and b.pending is not None
def test_target_exit_records_r_and_reason():
b = Broker(cfg())
b.place(short_order())
b.on_bar(12, t(12), 96.4, 96.6, 94.0, 94.5)
assert b.on_bar(13, t(13), 94.5, 94.8, 89.5, 90.5) == ["exit:target"]
tr = b.trades[0]
assert isinstance(tr, Trade)
assert (tr.side, tr.kind, tr.bias_id, tr.entry_bar, tr.exit_bar) == (-1, "RDIV", 7, 12, 13)
assert tr.entry_time == t(12) and tr.exit_time == t(13)
assert tr.risk == pytest.approx(3.02) and tr.r == pytest.approx(4.99 / 3.02)
assert tr.reason == "target" and tr.bars_held == 1
assert (tr.price_pnl, tr.direction, tr.comment) == (tr.r, -1, "RDIV")
assert b.position is None
def test_stop_exit_is_a_loss_of_about_one_r():
b = Broker(cfg())
b.place(short_order())
b.on_bar(12, t(12), 96.4, 96.6, 94.0, 94.5)
b.on_bar(13, t(13), 94.5, 98.5, 94.0, 98.0)
tr = b.trades[0]
assert tr.reason == "stop" and tr.exit == pytest.approx(98.01) and tr.r == pytest.approx(-1.0)
def test_same_bar_fill_and_stop_is_a_loss():
b = Broker(cfg())
b.place(short_order())
ev = b.on_bar(12, t(12), 96.4, 98.5, 94.0, 97.0)
assert ev == ["filled", "exit:stop"]
assert b.trades[0].reason == "stop" and b.trades[0].r == pytest.approx(-1.0)
def test_stop_and_target_in_one_bar_is_a_loss():
b = Broker(cfg())
b.place(short_order())
b.on_bar(12, t(12), 96.4, 96.6, 94.0, 94.5)
b.on_bar(13, t(13), 94.5, 98.5, 89.0, 92.0)
assert b.trades[0].reason == "stop"
def test_mfe_and_mae_are_in_r():
b = Broker(cfg())
b.place(short_order())
b.on_bar(12, t(12), 96.4, 96.6, 94.0, 94.5) # fill 94.99; low 94.0 -> mfe 0.99
b.on_bar(13, t(13), 94.5, 96.0, 93.0, 95.0) # high 96.0 -> mae 1.01; low 93 -> mfe 1.99
b.close_at(14, t(14), 95.0, "bias_end")
tr = b.trades[0]
assert tr.mfe_r == pytest.approx(1.99 / 3.02) and tr.mae_r == pytest.approx(1.01 / 3.02)
assert tr.reason == "bias_end" and tr.exit == pytest.approx(95.0) and tr.exit_bar == 14
def test_close_at_applies_slippage():
b = Broker(cfg(slippage_ticks=1))
b.place(short_order())
b.on_bar(12, t(12), 96.4, 96.6, 94.0, 94.5)
b.close_at(13, t(13), 95.0, "bias_end")
assert b.trades[0].exit == pytest.approx(95.01)
def test_pending_cancelled_when_the_pullback_extreme_is_taken_out():
b = Broker(cfg())
b.place(short_order())
assert b.on_bar(12, t(12), 97.0, 98.2, 96.0, 97.5) == ["cancel:extreme_taken"]
assert b.pending is None
def test_fill_wins_over_extreme_taken_on_the_same_bar():
b = Broker(cfg())
b.place(short_order(stop=98.5)) # extreme 98.0, stop 98.5
ev = b.on_bar(12, t(12), 96.0, 98.2, 94.0, 96.0) # broke the entry AND traded above 98.0 but not the stop
assert ev == ["filled"] and b.position is not None
def test_pending_expires_after_pending_max_bars():
b = Broker(cfg(pending_max_bars=2))
b.place(short_order(placed=11))
assert b.on_bar(12, t(12), 96.4, 96.6, 95.5, 96.0) == []
assert b.on_bar(13, t(13), 96.4, 96.6, 95.5, 96.0) == ["cancel:expired"] # 13 - 11 >= 2
assert b.pending is None
def test_place_replaces_a_pending_order():
b = Broker(cfg())
assert b.place(short_order()) is False
assert b.place(short_order(entry=95.49)) is True
assert b.pending.entry == pytest.approx(95.49)
def test_cancel_reports_whether_there_was_an_order():
b = Broker(cfg())
assert b.cancel("bias_end") is False
b.place(short_order())
assert b.cancel("bias_end") is True and b.pending is None
def test_time_stop_exits_at_the_close():
b = Broker(cfg(max_hold_bars=2))
b.place(short_order())
b.on_bar(12, t(12), 96.4, 96.6, 94.0, 94.5)
assert b.on_bar(13, t(13), 94.5, 95.0, 94.0, 94.8) == []
assert b.on_bar(14, t(14), 94.8, 95.2, 94.2, 95.0) == ["exit:time"]
assert b.trades[0].reason == "time" and b.trades[0].exit == pytest.approx(95.0)
def test_long_side_mirrors():
b = Broker(cfg())
b.place(long_order())
assert b.on_bar(12, t(12), 104.0, 106.0, 103.8, 105.5) == ["filled"]
assert b.position.fill == pytest.approx(105.01)
assert b.on_bar(13, t(13), 105.5, 110.5, 105.0, 110.0) == ["exit:target"]
assert b.trades[0].r == pytest.approx(4.99 / 3.02)
Run: cd "D:\vwap tpo pine\pyakao" && python -m pytest tests/test_entry5m_execution.py -q
Expected: FAIL with ModuleNotFoundError: No module named 'pyakao.entry5m.execution'.
execution.py"""Execution: one stop order, one position, R accounting (spec section 6).
Bar order inside `on_bar`: an open position is checked for stop, then target,
then the time stop; a pending order is then checked for a fill (with the
same-bar stop check), then for the pullback extreme being taken out, then for
expiry. Stops are checked before targets so a bar that hits both is a loss.
"""
from __future__ import annotations
from dataclasses import dataclass
from datetime import datetime
from .config import Entry5mConfig
from .setup import PendingOrder
EXIT_STOP = "stop"
EXIT_TARGET = "target"
EXIT_TIME = "time"
EXIT_BIAS_END = "bias_end"
@dataclass
class Position:
order: PendingOrder
fill: float
fill_bar: int
fill_time: datetime
mfe: float = 0.0 # best favourable excursion, price units
mae: float = 0.0 # worst adverse excursion, price units
@dataclass
class Trade:
side: int
kind: str
bias_id: int
entry_bar: int
exit_bar: int
entry_time: datetime
exit_time: datetime
entry: float
exit: float
stop: float
target: float
risk: float # |stop - order entry|, pre-slippage
r: float # (exit - entry) * side / risk
reason: str
mfe_r: float
mae_r: float
@property
def bars_held(self) -> int:
return self.exit_bar - self.entry_bar
# zms.metrics compatibility
@property
def price_pnl(self) -> float:
return self.r
@property
def direction(self) -> int:
return self.side
@property
def comment(self) -> str:
return self.kind
class Broker:
def __init__(self, cfg: Entry5mConfig):
self.cfg = cfg
self.pending: PendingOrder | None = None
self.position: Position | None = None
self.trades: list[Trade] = []
# ------------------------------------------------------------ orders ---
def place(self, order: PendingOrder) -> bool:
replaced = self.pending is not None
self.pending = order
return replaced
def cancel(self, reason: str) -> bool:
had = self.pending is not None
self.pending = None
return had
# -------------------------------------------------------------- bars ---
def on_bar(self, i: int, t: datetime, o: float, h: float, l: float, c: float) -> list[str]:
ev: list[str] = []
slip = self.cfg.slip
pos = self.position
if pos is not None:
self._track(pos, h, l)
reason = self._exit_reason(pos, h, l, i)
if reason is not None:
self._close(pos, i, t, self._exit_price(pos, reason, c), reason)
ev.append(f"exit:{reason}")
od = self.pending
if od is not None and self.position is None:
touched = (l <= od.entry) if od.side < 0 else (h >= od.entry)
if touched:
if od.side < 0:
fill = min(o, od.entry) - slip
else:
fill = max(o, od.entry) + slip
pos = Position(od, fill, i, t)
self.position = pos
self.pending = None
ev.append("filled")
self._track(pos, h, l)
hit_stop = (h >= od.stop) if od.side < 0 else (l <= od.stop)
if hit_stop:
self._close(pos, i, t, self._exit_price(pos, EXIT_STOP, c), EXIT_STOP)
ev.append(f"exit:{EXIT_STOP}")
else:
taken = (h > od.pb_extreme) if od.side < 0 else (l < od.pb_extreme)
if taken:
self.pending = None
ev.append("cancel:extreme_taken")
elif i - od.placed_bar >= self.cfg.pending_max_bars:
self.pending = None
ev.append("cancel:expired")
return ev
def close_at(self, i: int, t: datetime, price: float, reason: str) -> None:
pos = self.position
if pos is None:
return
slip = self.cfg.slip
px = price + slip if pos.order.side < 0 else price - slip # a short exits by buying
self._close(pos, i, t, px, reason)
# ----------------------------------------------------------- helpers ---
@staticmethod
def _track(pos: Position, h: float, l: float) -> None:
if pos.order.side < 0:
pos.mfe = max(pos.mfe, pos.fill - l)
pos.mae = max(pos.mae, h - pos.fill)
else:
pos.mfe = max(pos.mfe, h - pos.fill)
pos.mae = max(pos.mae, pos.fill - l)
def _exit_reason(self, pos: Position, h: float, l: float, i: int) -> str | None:
od = pos.order
if (h >= od.stop) if od.side < 0 else (l <= od.stop):
return EXIT_STOP
if (l <= od.target) if od.side < 0 else (h >= od.target):
return EXIT_TARGET
if self.cfg.max_hold_bars > 0 and i - pos.fill_bar >= self.cfg.max_hold_bars:
return EXIT_TIME
return None
def _exit_price(self, pos: Position, reason: str, close: float) -> float:
od = pos.order
slip = self.cfg.slip
if reason == EXIT_STOP:
return od.stop + slip if od.side < 0 else od.stop - slip
if reason == EXIT_TARGET:
return od.target
return close + slip if od.side < 0 else close - slip
def _close(self, pos: Position, i: int, t: datetime, px: float, reason: str) -> None:
od = pos.order
risk = abs(od.stop - od.entry)
r = (px - pos.fill) * od.side / risk
self.trades.append(Trade(
side=od.side, kind=od.kind, bias_id=od.bias_id,
entry_bar=pos.fill_bar, exit_bar=i, entry_time=pos.fill_time, exit_time=t,
entry=pos.fill, exit=px, stop=od.stop, target=od.target, risk=risk, r=r,
reason=reason, mfe_r=pos.mfe / risk, mae_r=pos.mae / risk,
))
self.position = None
Run: cd "D:\vwap tpo pine\pyakao" && python -m pytest tests/test_entry5m_execution.py -q
Expected: 17 passed.
cd "D:\vwap tpo pine" && git add pyakao/src/pyakao/entry5m/execution.py pyakao/tests/test_entry5m_execution.py
git commit -m "feat(entry5m): broker with stop-entry fills, exits and R accounting" -m "Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>"
Files:
pyakao/src/pyakao/entry5m/runner.pypyakao/src/pyakao/entry5m/__init__.pypyakao/tests/test_entry5m_runner.pyInterfaces:
Consumes: BiasTracker, Bias (Task 2); PullbackScanner (Task 3); Broker, Trade (Task 4); pyakao.zms.htf.build_htf(bars, tf, origin_offset_s) -> HtfSeries with .bars, .completed, .is_new(i); pyakao.data.Bars; pyakao.feeds.base.spec_for.
Produces:
Entry5mResult (dataclass): symbol: str, config: Entry5mConfig, trades: list[Trade], biases: list[Bias], counters: dict[str, int], n_bars: int, n_h1: int, start_time: datetime | None, end_time: datetime | None.run_entry5m(bars: Bars, cfg: Entry5mConfig | None = None, symbol: str = "", engine=None) -> Entry5mResult. When symbol is given and cfg does not override them, mintick/slippage_ticks come from spec_for(symbol). engine is passed to BiasTracker (tests only).biases, biases_RDIV, biases_HDIV, bias_end_target, bias_end_contradiction, bias_end_timeout, bias_end_replaced, bias_no_target, pullbacks_too_short, pullbacks_qualifying, rr_rejected, already_broken, skipped_in_position, orders_placed, cancel_extreme_taken, cancel_expired, cancel_bias_end, cancel_replaced, filled.[ ] Step 1: Write the failing tests (pyakao/tests/test_entry5m_runner.py)
"""run_entry5m: wiring, activation timing, one worked trade, no lookahead."""
import random
from datetime import datetime, timedelta, timezone
from types import SimpleNamespace
import pytest
from pyakao.data import Bars
from pyakao.entry5m import Entry5mConfig, Entry5mResult, run_entry5m
from pyakao.ztd import ZtdConfig
T0 = datetime(2024, 1, 1, tzinfo=timezone.utc)
NAN = float("nan")
def bars_from(ohlc):
times = [T0 + timedelta(minutes=5 * i) for i in range(len(ohlc))]
o, h, l, c = (list(x) for x in zip(*ohlc))
return Bars(times, o, h, l, c, [1.0] * len(ohlc))
def row(**kw):
base = dict(sigBear=0, sigBull=0, contraB=0, contraU=0, ext=0, curHi=NAN, curLo=NAN,
fRefB=NAN, fRefU=NAN, midLoB=NAN, midHiU=NAN)
base.update(kw)
return SimpleNamespace(**base)
class Scripted:
def __init__(self, rows):
self.rows, self.n = rows, 0
def update(self, *a, **k):
self.n += 1
return self.rows[self.n - 1] if self.n - 1 < len(self.rows) else row()
# H1 bars 0..3 flat (5m bars 0..47); ZTD fires a short bias on H1 bar 3, known at 5m bar 48.
FLAT = [(100.0, 100.5, 99.5, 100.0)] * 48
SCENARIO = FLAT + [
(100.0, 100.2, 99.0, 99.2), (99.2, 99.4, 98.0, 98.2), (98.2, 98.4, 97.0, 97.2), # 48..50
(97.2, 97.4, 96.0, 96.2), (96.2, 96.4, 95.0, 95.4), # 51..52 origin low 95 @52
(95.4, 96.5, 95.2, 96.3), (96.3, 97.2, 96.0, 97.0), (97.0, 97.6, 96.8, 97.4), # 53..55
(97.4, 98.0, 97.2, 97.8), # 56 pullback high 98
(97.8, 97.9, 97.0, 97.2), (97.2, 97.5, 96.6, 96.8), (96.8, 97.0, 96.2, 96.4), # 57..59 (confirm @59)
(96.4, 96.6, 94.0, 94.5), # 60 fill at 94.99
(94.5, 94.8, 89.5, 90.5), # 61 target 90 hit
(90.5, 91.0, 90.0, 90.8), (90.8, 91.2, 90.2, 91.0), # 62..63
]
FIRE = [row(), row(), row(), row(sigBear=2, fRefB=101.0, curHi=100.5, midLoB=90.0)]
def scenario_cfg(**kw):
base = dict(mintick=0.01, slippage_ticks=0)
base.update(kw)
return Entry5mConfig(**base)
def test_worked_short_trade_end_to_end():
res = run_entry5m(bars_from(SCENARIO), scenario_cfg(), engine=Scripted(FIRE))
assert isinstance(res, Entry5mResult)
assert res.n_bars == len(SCENARIO) and res.n_h1 == 5
(b,) = res.biases
assert (b.side, b.start_h1, b.start_5m, b.target) == (-1, 3, 48, 90.0)
assert b.end_reason == "target" and b.end_5m == 61
(tr,) = res.trades
assert (tr.entry_bar, tr.exit_bar, tr.reason) == (60, 61, "target")
assert tr.entry == pytest.approx(94.99) and tr.stop == pytest.approx(98.01) and tr.exit == 90.0
assert tr.r == pytest.approx(4.99 / 3.02)
c = res.counters
assert c["biases"] == 1 and c["biases_RDIV"] == 1 and c["bias_end_target"] == 1
assert c["pullbacks_qualifying"] == 1 and c["orders_placed"] == 1 and c["filled"] == 1
assert c["cancel_bias_end"] == 0 and c["skipped_in_position"] == 0
def test_bias_end_from_h1_closes_the_position_and_cancels_the_order():
# same scenario, but the H1 bar closing at 5m bar 60 (H1 bar 4 = bars 48..59) contradicts
rows = FIRE + [row(contraB=1)]
res = run_entry5m(bars_from(SCENARIO), scenario_cfg(), engine=Scripted(rows))
(b,) = res.biases
assert b.end_reason == "contradiction" and b.end_5m == 60
(tr,) = res.trades # filled at bar 60 then closed at bar 60's close
assert (tr.entry_bar, tr.exit_bar, tr.reason) == (60, 60, "bias_end") and tr.exit == pytest.approx(94.5)
assert res.counters["bias_end_contradiction"] == 1
def test_pending_order_is_cancelled_when_the_bias_ends():
rows = FIRE + [row(contraB=1)]
ohlc = SCENARIO[:60] + [(96.4, 96.6, 95.5, 96.0)] * 4 # bar 60 does not fill
res = run_entry5m(bars_from(ohlc), scenario_cfg(), engine=Scripted(rows))
assert res.trades == [] and res.counters["orders_placed"] == 1 and res.counters["cancel_bias_end"] == 1
def test_pullbacks_while_in_position_are_skipped():
ohlc = SCENARIO[:61] + [
(94.5, 95.0, 94.2, 94.8), (94.8, 95.6, 94.6, 95.4), (95.4, 96.0, 95.2, 95.8), # 61..63 rise
(95.8, 96.4, 95.6, 96.2), # 64 pivot high 96.4
(96.2, 96.3, 95.4, 95.6), (95.6, 95.9, 95.0, 95.2), (95.2, 95.5, 94.6, 94.8), # 65..67 confirm @67
(94.8, 95.0, 94.0, 94.2), (94.2, 94.4, 93.6, 93.8),
]
res = run_entry5m(bars_from(ohlc), scenario_cfg(), engine=Scripted(FIRE))
assert res.trades == [] or res.trades[0].reason != "target"
assert res.counters["skipped_in_position"] == 1
def test_symbol_spec_sets_costs_unless_overridden():
res = run_entry5m(bars_from(FLAT), Entry5mConfig(), symbol="XAUUSD", engine=Scripted([]))
assert res.config.mintick == 0.01 and res.config.slippage_ticks == 18
res = run_entry5m(bars_from(FLAT), Entry5mConfig(slippage_ticks=5), symbol="XAUUSD", engine=Scripted([]))
assert res.config.slippage_ticks == 5
def _walk(n, seed):
rng = random.Random(seed)
px = 100.0
ohlc = []
for _ in range(n):
o = px
moves = [rng.gauss(0, 0.05) for _ in range(4)]
path = [o + sum(moves[:k + 1]) for k in range(4)]
h, l, c = max(o, *path), min(o, *path), path[-1]
ohlc.append((o, h, l, c))
px = c
return bars_from(ohlc)
def _fast():
return Entry5mConfig(mintick=0.01, slippage_ticks=1, ztd=ZtdConfig(
t_thr_mode="Fixed sigma", o_thr_mode="Fixed sigma",
t_z_elev=0.5, t_z_ext=1.0, o_z_elev=0.5, o_z_ext=1.0, sigma_len=5))
def test_random_walk_produces_trades_with_the_fast_ztd():
res = run_entry5m(_walk(9000, 7), _fast())
assert res.counters["biases"] > 0 and len(res.trades) > 0
def test_no_lookahead_prefix_property():
full = run_entry5m(_walk(9000, 7), _fast())
cut = 6000
part = run_entry5m(_walk(6000, 7), _fast())
done_full = [(t.entry_bar, t.exit_bar, t.entry, t.exit, t.reason) for t in full.trades if t.exit_bar < cut - 1]
done_part = [(t.entry_bar, t.exit_bar, t.entry, t.exit, t.reason) for t in part.trades if t.exit_bar < cut - 1]
assert done_full == done_part
starts_full = [(b.start_5m, b.side, b.target) for b in full.biases if b.start_5m < cut - 1]
starts_part = [(b.start_5m, b.side, b.target) for b in part.biases if b.start_5m < cut - 1]
assert starts_full == starts_part
Run: cd "D:\vwap tpo pine\pyakao" && python -m pytest tests/test_entry5m_runner.py -q
Expected: FAIL with ImportError: cannot import name 'run_entry5m'.
runner.py"""One pass over the 5m bars: bias (H1) -> pullbacks -> orders -> fills.
Per 5m bar i, in this order:
1. broker.on_bar intrabar fills and stop/target/time exits
2. H1 events if a new H1 bar closed before bar i opened: ends first
(close the position at bar i's close, cancel the order),
then a start (bias active from bar i)
3. target touch bias ends when bar i trades through the target
4. pullback scan a pivot confirmed on bar i may price an order for bar i+1
"""
from __future__ import annotations
from dataclasses import dataclass, field
from datetime import datetime
from ..data import Bars
from ..feeds.base import spec_for
from ..zms.htf import build_htf
from .bias import Bias, BiasTracker, END_TARGET
from .config import Entry5mConfig
from .execution import Broker, Trade, EXIT_BIAS_END
from .setup import PullbackScanner
COUNTER_KEYS = (
"biases", "biases_RDIV", "biases_HDIV",
"bias_end_target", "bias_end_contradiction", "bias_end_timeout", "bias_end_replaced", "bias_no_target",
"pullbacks_too_short", "pullbacks_qualifying", "rr_rejected", "already_broken", "skipped_in_position",
"orders_placed", "cancel_extreme_taken", "cancel_expired", "cancel_bias_end", "cancel_replaced", "filled",
)
@dataclass
class Entry5mResult:
symbol: str
config: Entry5mConfig
trades: list[Trade] = field(default_factory=list)
biases: list[Bias] = field(default_factory=list)
counters: dict[str, int] = field(default_factory=lambda: {k: 0 for k in COUNTER_KEYS})
n_bars: int = 0
n_h1: int = 0
start_time: datetime | None = None
end_time: datetime | None = None
def run_entry5m(bars: Bars, cfg: Entry5mConfig | None = None, symbol: str = "",
engine=None) -> Entry5mResult:
cfg = cfg or Entry5mConfig()
if symbol:
spec = spec_for(symbol)
defaults = Entry5mConfig()
if cfg.mintick == defaults.mintick:
cfg.mintick = spec.mintick
if cfg.slippage_ticks == defaults.slippage_ticks:
cfg.slippage_ticks = spec.slippage_ticks
res = Entry5mResult(symbol=symbol, config=cfg, n_bars=len(bars))
if len(bars) == 0:
return res
res.start_time, res.end_time = bars.times[0], bars.times[-1]
cnt = res.counters
htf = build_htf(bars, cfg.htf, cfg.htf_offset_s)
tracker = BiasTracker(cfg, engine=engine)
scanner = PullbackScanner(cfg)
broker = Broker(cfg)
def bias_ended(b: Bias, i: int, c: float, t: datetime) -> None:
b.end_5m = i
cnt["bias_end_" + b.end_reason] += 1
if broker.position is not None:
broker.close_at(i, t, c, EXIT_BIAS_END)
if broker.cancel("bias_end"):
cnt["cancel_bias_end"] += 1
scanner.clear_bias()
for i in range(len(bars)):
t, o, h, l, c = bars.times[i], bars.opens[i], bars.highs[i], bars.lows[i], bars.closes[i]
for ev in broker.on_bar(i, t, o, h, l, c):
if ev == "filled":
cnt["filled"] += 1
elif ev.startswith("cancel:"):
cnt["cancel_" + ev.split(":", 1)[1]] += 1
if htf.is_new(i):
res.n_h1 += 1
for kind, b in tracker.on_h1_close(htf.bars[htf.completed[i]]):
if kind == "end":
bias_ended(b, i, c, t)
else:
b.start_5m = i
cnt["biases"] += 1
cnt["biases_" + b.kind] += 1
scanner.set_bias(b.side, i)
b = tracker.active
if b is not None and ((l <= b.target) if b.side < 0 else (h >= b.target)):
tracker.end_active(END_TARGET, i)
bias_ended(b, i, c, t)
pb, why = scanner.on_bar(i, h, l)
if why == "too_short":
cnt["pullbacks_too_short"] += 1
if pb is not None and tracker.active is not None:
cnt["pullbacks_qualifying"] += 1
if broker.position is not None:
cnt["skipped_in_position"] += 1
else:
order, oreason = scanner.order_for(pb, tracker.active)
if order is None:
cnt[oreason] += 1
else:
if broker.place(order):
cnt["cancel_replaced"] += 1
cnt["orders_placed"] += 1
res.trades = broker.trades
res.biases = tracker.history + ([tracker.active] if tracker.active else [])
cnt["bias_no_target"] = tracker.no_target
return res
Update __init__.py:
from .config import Entry5mConfig
from .runner import Entry5mResult, run_entry5m
__all__ = ["Entry5mConfig", "Entry5mResult", "run_entry5m"]
Run: cd "D:\vwap tpo pine\pyakao" && python -m pytest tests/test_entry5m_runner.py -q
Expected: 7 passed. If test_random_walk_produces_trades_with_the_fast_ztd finds zero biases, change the seed in BOTH random-walk tests (they must share it) until it does, and record the seed in the report; do not change the config.
Run: cd "D:\vwap tpo pine\pyakao" && python -m pytest -q — expected all green (208 + 14 + 11 + 17 + 7 = 257).
cd "D:\vwap tpo pine" && git add pyakao/src/pyakao/entry5m/runner.py pyakao/src/pyakao/entry5m/__init__.py pyakao/tests/test_entry5m_runner.py
git commit -m "feat(entry5m): runner wiring bias, pullbacks and broker on the 5m clock" -m "Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>"
Files:
pyakao/src/pyakao/entry5m/report.py, pyakao/src/pyakao/entry5m/screen.pypyakao/tests/test_entry5m_report.pyInterfaces:
Consumes: Entry5mResult, run_entry5m, Entry5mConfig; Trade; pyakao.feeds.tickstory.load_tickstory.
Produces:
summarize(res: Entry5mResult) -> dict with keys n, wins, losses, win_rate, net_r, expectancy, pf, ex_outlier_pf, max_dd_r, best_month, best_month_share, first_half_r, second_half_r, avg_bars_held plus every counter.judge(summary: dict) -> dict with n_ok, pf_ok, expectancy_ok, month_ok, halves_ok, passed (bools).GROUPS = {"usd": ("EURUSD", "GBPUSD"), "metal": ("XAUUSD",), "jpy": ("EURJPY",)}; verdict(summaries: dict[str, dict]) -> dict with per-group pass and overall passed (>= 2 groups).format_table(summaries, judgements, verdict) -> str; format_sensitivity(grid: dict[tuple[int, float], dict[str, dict]]) -> str.screen.main(argv) -> int; CLI flags --symbols, --root, --tf, --out, --no-sensitivity.[ ] Step 1: Write the failing tests (pyakao/tests/test_entry5m_report.py)
"""Report: the pre-registered table and verdict (spec sections 7-8)."""
from datetime import datetime, timedelta, timezone
import pytest
from pyakao.entry5m import Entry5mConfig, Entry5mResult
from pyakao.entry5m.execution import Trade
from pyakao.entry5m.report import GROUPS, format_table, judge, summarize, verdict
T0 = datetime(2024, 1, 1, tzinfo=timezone.utc)
def trade(r, day, side=-1, held=3):
t = T0 + timedelta(days=day)
return Trade(side, "RDIV", 1, 0, held, t, t + timedelta(minutes=5 * held), 100.0, 100.0 - r * side,
101.0, 95.0, 1.0, r, "target" if r > 0 else "stop", max(r, 0.0), max(-r, 0.0))
def result(rs, days=None):
res = Entry5mResult("EURUSD", Entry5mConfig())
res.trades = [trade(r, d) for r, d in zip(rs, days or range(len(rs)))]
res.start_time, res.end_time = T0, T0 + timedelta(days=max(days or [len(rs)]) + 1)
res.counters["biases"] = 3
return res
def test_summary_metrics():
s = summarize(result([2.0, -1.0, 3.0, -1.0, 1.0], days=[0, 1, 2, 40, 41]))
assert (s["n"], s["wins"], s["losses"]) == (5, 3, 2)
assert s["net_r"] == pytest.approx(4.0) and s["expectancy"] == pytest.approx(0.8)
assert s["pf"] == pytest.approx(6.0 / 2.0)
assert s["ex_outlier_pf"] == pytest.approx(0.0 / 2.0) # drop the three best winners: none left
assert s["max_dd_r"] == pytest.approx(1.0)
assert s["best_month"] == "2024-01" and s["best_month_share"] == pytest.approx(100.0)
assert s["first_half_r"] == pytest.approx(4.0) and s["second_half_r"] == pytest.approx(0.0)
assert s["avg_bars_held"] == pytest.approx(3.0) and s["biases"] == 3
def test_ex_outlier_drops_exactly_three_winners():
s = summarize(result([5.0, 4.0, 3.0, 2.0, 1.0, -1.0, -1.0]))
assert s["ex_outlier_pf"] == pytest.approx(3.0 / 2.0)
def test_empty_result_summarizes_without_errors():
s = summarize(Entry5mResult("EURUSD", Entry5mConfig()))
assert s["n"] == 0 and s["net_r"] == 0.0 and s["best_month"] is None
assert judge(s)["passed"] is False
def test_judge_applies_the_pre_registered_thresholds():
good = dict(n=200, pf=1.16, ex_outlier_pf=1.01, expectancy=0.051, best_month_share=35.0,
first_half_r=1.0, second_half_r=1.0, net_r=2.0)
assert judge(good)["passed"] is True
for k, v in [("n", 199), ("pf", 1.15), ("ex_outlier_pf", 1.0), ("expectancy", 0.05),
("best_month_share", 35.1), ("first_half_r", 0.0), ("second_half_r", -0.1)]:
bad = dict(good, **{k: v})
assert judge(bad)["passed"] is False, k
def test_verdict_needs_two_groups_and_gbpusd_not_negative():
ok = dict(n=200, pf=1.2, ex_outlier_pf=1.1, expectancy=0.1, best_month_share=20.0,
first_half_r=1.0, second_half_r=1.0, net_r=2.0)
bad = dict(ok, pf=0.9, ex_outlier_pf=0.8, expectancy=-0.1, net_r=-2.0)
v = verdict({"EURUSD": ok, "GBPUSD": ok, "XAUUSD": bad, "EURJPY": ok})
assert v["groups"] == {"usd": True, "metal": False, "jpy": True} and v["passed"] is True
v = verdict({"EURUSD": ok, "GBPUSD": bad, "XAUUSD": bad, "EURJPY": ok})
assert v["groups"]["usd"] is False and v["passed"] is False
v = verdict({"EURUSD": ok, "GBPUSD": ok, "XAUUSD": bad, "EURJPY": bad})
assert v["passed"] is False
assert set(GROUPS) == {"usd", "metal", "jpy"}
def test_table_mentions_every_symbol_and_the_verdict():
s = {"EURUSD": summarize(result([1.0, -1.0])), "XAUUSD": summarize(result([2.0]))}
j = {k: judge(v) for k, v in s.items()}
txt = format_table(s, j, verdict(s))
assert "EURUSD" in txt and "XAUUSD" in txt and ("PASS" in txt or "FAIL" in txt)
Run: cd "D:\vwap tpo pine\pyakao" && python -m pytest tests/test_entry5m_report.py -q
Expected: FAIL with ModuleNotFoundError: No module named 'pyakao.entry5m.report'.
report.py"""The pre-registered table and verdict (spec sections 7 and 8).
Everything here is judged in R. The thresholds are the ones committed in the
spec before any five-year run; they are constants, not parameters.
"""
from __future__ import annotations
from datetime import datetime, timezone
from .runner import Entry5mResult
MIN_TRADES = 200
MIN_PF = 1.15
MIN_EX_OUTLIER_PF = 1.0
MIN_EXPECTANCY = 0.05
MAX_BEST_MONTH_SHARE = 35.0
GROUPS: dict[str, tuple[str, ...]] = {"usd": ("EURUSD", "GBPUSD"), "metal": ("XAUUSD",), "jpy": ("EURJPY",)}
MIN_GROUPS = 2
def summarize(res: Entry5mResult) -> dict:
rs = [t.r for t in res.trades]
wins = [r for r in rs if r > 0]
losses = [r for r in rs if r <= 0]
gross_win, gross_loss = sum(wins), -sum(losses)
net = sum(rs)
n = len(rs)
def pf(gw, gl):
if gl > 0:
return gw / gl
return float("inf") if gw > 0 else float("nan")
ex_win = sum(sorted(wins, reverse=True)[3:])
running = peak = dd = 0.0
for r in rs:
running += r
peak = max(peak, running)
dd = max(dd, peak - running)
months: dict[str, float] = {}
for t in res.trades:
k = f"{t.exit_time.year:04d}-{t.exit_time.month:02d}"
months[k] = months.get(k, 0.0) + t.r
if months:
best = max(months, key=lambda k: months[k])
share = 100.0 * months[best] / net if net > 0 else float("nan")
else:
best, share = None, float("nan")
if res.start_time and res.end_time:
mid = res.start_time + (res.end_time - res.start_time) / 2
else:
mid = datetime.max.replace(tzinfo=timezone.utc)
first = sum(t.r for t in res.trades if t.exit_time < mid)
second = sum(t.r for t in res.trades if t.exit_time >= mid)
out = {
"symbol": res.symbol,
"n": n, "wins": len(wins), "losses": len(losses),
"win_rate": 100.0 * len(wins) / n if n else float("nan"),
"net_r": net, "expectancy": net / n if n else float("nan"),
"pf": pf(gross_win, gross_loss), "ex_outlier_pf": pf(ex_win, gross_loss),
"max_dd_r": dd, "best_month": best, "best_month_share": share,
"first_half_r": first, "second_half_r": second,
"avg_bars_held": (sum(t.bars_held for t in res.trades) / n) if n else float("nan"),
"n_bars": res.n_bars, "n_h1": res.n_h1,
}
out.update(res.counters)
return out
def judge(s: dict) -> dict:
def ok(x, thr, above=True):
return (x == x) and ((x > thr) if above else (x <= thr))
j = {
"n_ok": s["n"] >= MIN_TRADES,
"pf_ok": ok(s["pf"], MIN_PF) and ok(s["ex_outlier_pf"], MIN_EX_OUTLIER_PF),
"expectancy_ok": ok(s["expectancy"], MIN_EXPECTANCY),
"month_ok": ok(s["best_month_share"], MAX_BEST_MONTH_SHARE, above=False),
"halves_ok": s["first_half_r"] > 0 and s["second_half_r"] > 0,
}
j["passed"] = all(j.values())
return j
def verdict(summaries: dict[str, dict]) -> dict:
groups: dict[str, bool] = {}
for g, syms in GROUPS.items():
lead = syms[0]
if lead not in summaries:
groups[g] = False
continue
ok = judge(summaries[lead])["passed"]
for other in syms[1:]:
if other in summaries and summaries[other]["net_r"] < 0:
ok = False
groups[g] = ok
n_pass = sum(groups.values())
return {"groups": groups, "groups_passed": n_pass, "passed": n_pass >= MIN_GROUPS}
def _f(x, nd=2):
if x is None or x != x:
return "-"
if x == float("inf"):
return "inf"
return f"{x:.{nd}f}"
def format_table(summaries: dict[str, dict], judgements: dict[str, dict], v: dict) -> str:
cols = [("symbol", 8), ("n", 5), ("win%", 6), ("netR", 8), ("expR", 7), ("PF", 6), ("exPF", 6),
("ddR", 7), ("month", 8), ("share%", 7), ("h1R", 8), ("h2R", 8), ("bars", 6), ("verdict", 7)]
lines = [" ".join(name.rjust(w) for name, w in cols)]
for sym in sorted(summaries):
s, j = summaries[sym], judgements[sym]
vals = [sym, str(s["n"]), _f(s["win_rate"], 1), _f(s["net_r"], 1), _f(s["expectancy"], 3),
_f(s["pf"]), _f(s["ex_outlier_pf"]), _f(s["max_dd_r"], 1), s["best_month"] or "-",
_f(s["best_month_share"], 1), _f(s["first_half_r"], 1), _f(s["second_half_r"], 1),
_f(s["avg_bars_held"], 1), "PASS" if j["passed"] else "FAIL"]
lines.append(" ".join(val.rjust(w) for val, (_, w) in zip(vals, cols)))
lines.append("")
lines.append("bias / pullback funnel:")
keys = ["biases", "biases_RDIV", "biases_HDIV", "bias_end_target", "bias_end_contradiction",
"bias_end_timeout", "bias_end_replaced", "bias_no_target", "pullbacks_too_short",
"pullbacks_qualifying", "rr_rejected", "already_broken", "skipped_in_position", "orders_placed",
"cancel_extreme_taken", "cancel_expired", "cancel_bias_end", "cancel_replaced", "filled"]
for sym in sorted(summaries):
s = summaries[sym]
lines.append(f" {sym:8s} " + " ".join(f"{k}={s[k]}" for k in keys))
lines.append("")
for g, okg in v["groups"].items():
lines.append(f" {'PASS' if okg else 'FAIL'} group {g} ({', '.join(GROUPS[g])})")
lines.append(f" {'PASS' if v['passed'] else 'FAIL'} overall: {v['groups_passed']} of {len(GROUPS)} groups "
f"(need {MIN_GROUPS})")
return "\n".join(lines)
def format_sensitivity(grid: dict[tuple[int, float], dict[str, dict]]) -> str:
"""grid[(pivot_n, min_rr)][symbol] -> summary. Reported, never judged."""
syms = sorted({s for cell in grid.values() for s in cell})
lines = ["sensitivity (NOT judged): net R / PF / n per cell",
"cell " + " ".join(f"{s:>22s}" for s in syms)]
for (n, rr) in sorted(grid):
cells = []
for s in syms:
v = grid[(n, rr)].get(s)
cells.append(f"{_f(v['net_r'], 1):>8s} {_f(v['pf']):>6s} {v['n']:>6d}" if v else " " * 22)
lines.append(f"N={n} rr={rr:<4}" + " ".join(cells))
return "\n".join(lines)
screen.py"""CLI: run the primary cell (and the sensitivity grid) on Tickstory data.
python -m pyakao.entry5m.screen --symbols EURUSD,GBPUSD,XAUUSD,EURJPY --out ../entry5m/research/run.md
Writes the table to stdout and, with --out, to a Markdown file (plus a JSON
sidecar with every summary) so the run is reproducible from the file alone.
"""
from __future__ import annotations
import argparse
import json
import time
from pathlib import Path
from ..feeds.tickstory import load_tickstory
from ..ztd import ZtdEngine
from .bias import RecordingEngine, ReplayEngine
from .config import Entry5mConfig
from .report import format_sensitivity, format_table, judge, summarize, verdict
from .runner import run_entry5m
DEFAULT_ROOT = "D:/tickstory/History data"
DEFAULT_SYMBOLS = "EURUSD,GBPUSD,XAUUSD,EURJPY"
SENS_N = (2, 3, 5)
SENS_RR = (1.0, 1.5)
def load(root: str, symbol: str, tf: str):
return load_tickstory(Path(root) / f"{symbol}_mt5_bars.csv", timeframe=tf)
def main(argv=None) -> int:
ap = argparse.ArgumentParser(prog="python -m pyakao.entry5m.screen")
ap.add_argument("--symbols", default=DEFAULT_SYMBOLS)
ap.add_argument("--root", default=DEFAULT_ROOT)
ap.add_argument("--tf", default="5m", help="chart timeframe to resample the minute bars to (5m or 15m)")
ap.add_argument("--out", default="", help="Markdown file to write; a .json sidecar is written beside it")
ap.add_argument("--no-sensitivity", action="store_true")
a = ap.parse_args(argv)
symbols = [s.strip().upper() for s in a.symbols.split(",") if s.strip()]
bars = {}
for sym in symbols:
t0 = time.time()
bars[sym] = load(a.root, sym, a.tf)
print(f"loaded {sym}: {len(bars[sym])} {a.tf} bars in {time.time() - t0:.1f}s")
summaries, judgements, ztd_rows = {}, {}, {}
for sym in symbols:
t0 = time.time()
rec = RecordingEngine(ZtdEngine(Entry5mConfig().ztd)) # H1 ZTD computed once per symbol
res = run_entry5m(bars[sym], Entry5mConfig(), symbol=sym, engine=rec)
ztd_rows[sym] = rec.rows
summaries[sym] = summarize(res)
judgements[sym] = judge(summaries[sym])
print(f"ran {sym}: {summaries[sym]['n']} trades in {time.time() - t0:.1f}s")
v = verdict(summaries)
table = format_table(summaries, judgements, v)
grid = {}
if not a.no_sensitivity:
for n in SENS_N:
for rr in SENS_RR:
cell = {}
for sym in symbols:
cfg = Entry5mConfig(pivot_n=n, min_rr=rr)
cell[sym] = summarize(run_entry5m(bars[sym], cfg, symbol=sym,
engine=ReplayEngine(ztd_rows[sym])))
grid[(n, rr)] = cell
sens = format_sensitivity(grid) if grid else ""
text = "\n".join([f"# entry5m screen tf={a.tf} symbols={','.join(symbols)}", "", "```", table, "```", ""]
+ (["```", sens, "```", ""] if sens else []))
print(table)
if sens:
print(sens)
if a.out:
out = Path(a.out)
out.parent.mkdir(parents=True, exist_ok=True)
out.write_text(text, encoding="utf-8")
side = {"tf": a.tf, "summaries": summaries, "judgements": judgements, "verdict": v,
"sensitivity": {f"N={n} rr={rr}": cell for (n, rr), cell in grid.items()}}
out.with_suffix(".json").write_text(json.dumps(side, indent=1, default=str), encoding="utf-8")
print(f"wrote {out} and {out.with_suffix('.json')}")
return 0
if __name__ == "__main__":
raise SystemExit(main())
(The package __init__ must NOT import screen, so python -m pyakao.entry5m.screen stays free of the runpy warning. The sensitivity grid replays the recorded H1 ZTD rows: the grid only varies pivot_n and min_rr, which do not touch ZTD, and the H1 bar stream is identical across runs of the same symbol.)
Run: cd "D:\vwap tpo pine\pyakao" && python -m pytest tests/test_entry5m_report.py -q — expected 6 passed.
Smoke (real data, one symbol, no sensitivity, to a scratch file; record the wall time):
cd "D:\vwap tpo pine\pyakao" && python -m pyakao.entry5m.screen --symbols EURUSD --no-sensitivity --out "C:/Users/DANIEL~1/AppData/Local/Temp/claude/D--vwap-tpo-pine/b151af76-39c8-4c23-ae9d-1bae5fbb8ec5/scratchpad/entry5m_smoke.md"
Expected: loads ~370k bars, runs in under a few minutes, prints a table. Do not read the numbers as a result and do not change any parameter because of them; this step only proves the pipeline runs. Report the wall time.
cd "D:\vwap tpo pine" && git add pyakao/src/pyakao/entry5m/report.py pyakao/src/pyakao/entry5m/screen.py pyakao/tests/test_entry5m_report.py
git commit -m "feat(entry5m): pre-registered report, verdict and screen CLI" -m "Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>"
Files:
entry5m/research/2026-09-06-entry5m-screen.md (+ .json sidecar), entry5m/README.mdREADME.md (repo root; script/sub-project table), entry5m/docs/2026-09-06-entry5m-design.md (Status line only)C:\Users\Danielshobhan\.claude\projects\D--vwap-tpo-pine\memory\project-state.md (one paragraph, absolute dates)Interfaces: consumes the CLI from Task 6 only.
Run: cd "D:\vwap tpo pine" && git log --oneline -- entry5m/docs/2026-09-06-entry5m-design.md | tail -1
Expected: the spec commit (59d20bc or earlier) precedes every pyakao/src/pyakao/entry5m commit. Quote both hashes in the run record.
cd "D:\vwap tpo pine\pyakao" && python -m pyakao.entry5m.screen --symbols EURUSD,GBPUSD,XAUUSD,EURJPY --out ../entry5m/research/2026-09-06-entry5m-screen.md
Then the secondary 15m report:
cd "D:\vwap tpo pine\pyakao" && python -m pyakao.entry5m.screen --symbols EURUSD,GBPUSD,XAUUSD,EURJPY --tf 15m --no-sensitivity --out ../entry5m/research/2026-09-06-entry5m-screen-15m.md
Prepend to entry5m/research/2026-09-06-entry5m-screen.md a header with: the spec commit hash and the code HEAD hash; the primary cell values; the data window per symbol (first/last bar time from the JSON n_bars, or print them); the overall verdict line copied from the table (PASS/FAIL, groups passed); one paragraph per symbol reading the funnel counters (how many biases, how they ended, how many pullbacks became orders, fills), written as observations, not adjustments. If the verdict is FAIL, write the sentence "The rule as specified is dead; no parameter search was run (spec section 8)." If it is PASS, write "Passed the pre-registered criteria; next is the Pine strategy spec (spec section 10)." The sensitivity block stays labelled NOT judged.
entry5m/README.md: purpose (two sentences), pointers to the spec, the plan, the run record; the CLI command; the module map (one line per module); the verdict line. Spec Status: line becomes built and run 2026-09-06; verdict: <PASS|FAIL> (see research/2026-09-06-entry5m-screen.md). Root README.md: add a row/paragraph for entry5m/ next to the harmonic/ entry with the one-line verdict.
Memory (project-state.md): replace the "entry5m started" paragraph with the outcome: verdict, trade counts per symbol, where the record lives, and what is next (Pine strategy only on PASS; otherwise "entry rule dead as specified, Daniel decides the next variant").
cd "D:\vwap tpo pine" && git add entry5m/research/2026-09-06-entry5m-screen.md entry5m/research/2026-09-06-entry5m-screen.json entry5m/research/2026-09-06-entry5m-screen-15m.md entry5m/research/2026-09-06-entry5m-screen-15m.json entry5m/README.md README.md entry5m/docs/2026-09-06-entry5m-design.md
git commit -m "research(entry5m): pre-registered 5y screen on EURUSD/GBPUSD/XAUUSD/EURJPY; verdict recorded" -m "Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>"
Spec coverage. Section 3 data: Task 6 loader + Task 7 run (EURJPY control, EURGBP excluded). Section 4 bias: Task 1 (port fields) + Task 2 (tracker) + Task 5 (5m activation and target end). Section 5 setup: Task 3 (pivots, pullback, order, already-broken, rr floor), Task 4/5 (cancel reasons: extreme taken and expiry in the broker, bias-end and replaced in the runner), Task 5 (one position, skipped in position). Section 6 execution: Task 4 (fills, slip, same-bar conflicts, exits, R/MFE/MAE). Section 7 report: Task 6 (funnel, metrics, sensitivity, --tf 15m). Section 8 criteria: Task 6 constants + Task 7 run with the no-search rule. Section 9 testing: each task's tests; no-lookahead prefix test in Task 5; golden test untouched in meaning (Task 1). Section 10 Pine strategy: out of scope, noted in Task 7's record. Section 11 limits: gap fill rule (Task 4), runtime check (Task 6 smoke).
Placeholder scan. None; every code step is complete. The only conditional instructions are the seed note in Task 5 and the mirror-expression note in Task 3, both with the exact replacement given.
Type consistency. Bias fields identical in Tasks 2/3/5; PendingOrder field order (side, entry, stop, target, placed_bar, pb_bar, pb_extreme, origin_bar, bias_id, kind) used positionally in Task 3's order_for, Task 4's tests and Task 5; Trade positional order in Task 6's test matches Task 4's dataclass; counter keys in Task 5 COUNTER_KEYS match Task 6's format_table list and the reasons returned by order_for (rr_rejected, already_broken) and the broker events (cancel:extreme_taken, cancel:expired); END_TARGET/EXIT_BIAS_END imported where used; summarize keys match judge, verdict, format_table and the report tests.