diff --git a/Dependencies/env.example b/Dependencies/env.example index b5c0ddd..bb9d37a 100644 --- a/Dependencies/env.example +++ b/Dependencies/env.example @@ -140,6 +140,25 @@ SESSION_STATE_SNAPSHOT_SECONDS=30.0 # authority. Every replaced state file is archived beside the configured file as # `session_state..recovery.json` before the new run writes anything. SESSION_STATE_RESUME_ENABLED=false +# How often the supervisor logs a liveness line (workers alive, marks writes, +# marks age, open positions). MainThread is otherwise silent through a healthy +# session, so when the snapshot loop stopped on 2026-08-12 its silence carried no +# information and the freeze was only found hours later from a file mtime. At the +# default that is ~75 lines a session, and a stall becomes visible within minutes +# instead of at end of day. Raise it to quieten the log; do not disable it. +SESSION_STATE_HEARTBEAT_SECONDS=300.0 +# Persist trade events from a BACKGROUND writer instead of on the trading thread +# that published them. OFF by default: it trades a small amount of the durability +# ADR-0012 exists to provide -- a hard kill can lose events queued in the last +# write cycle, where the synchronous path loses only the one in flight. +# +# Turn it ON if disk stalls are costing you the market feed. On 2026-08-11, 86% +# of websocket disconnects (25 of 29) landed within 20s of a session-state write +# stall, with 'keepalive ping timeout' dominant -- a blocked event loop drops the +# feed. The data reaches the platter at the same moment either way; what changes +# is that no trading thread waits for it. Lost queued events still have their +# EXIT log line, which is the source the EOD Sheet parses anyway. +SESSION_STATE_ASYNC_WRITES=false # Underlying instrument symbol prefix used to filter the option chain. UNDERLYING=NIFTY diff --git a/Dependencies/market_data_health.py b/Dependencies/market_data_health.py index 8d32935..e0c2db7 100644 --- a/Dependencies/market_data_health.py +++ b/Dependencies/market_data_health.py @@ -237,6 +237,40 @@ def newest_completed_minute_timestamp( return max(completed, default=None) +def stable_official_minutes( + validated_frame: pd.DataFrame, + *, + request_started_at: datetime, + grace_seconds: float, +) -> frozenset[pd.Timestamp]: + """Return REST minute stamps safely final at the request's start time. + + Dhan may include the currently forming candle in a REST response. A slow + HTTP response must not turn that provisional row into official evidence, + because the request might have started before the candle had closed. The + request-start clock is therefore authoritative: a row is usable only when + its start stamp is strictly before ``floor(request_started_at - grace)``. + + ``validated_frame`` is the normalized output of :func:`validate_ohlc_frame`. + Its timestamps are usually naive IST, but this helper deliberately accepts + aware values too so future adapters keep the same safety boundary. Empty + frames remain empty because callers may use this pure helper before the + store's non-empty publication validation. + """ + + if validated_frame is None or validated_frame.empty or "timestamp" not in validated_frame: + return frozenset() + request_ist = _as_aware_ist(request_started_at) + boundary = pd.Timestamp(request_ist - timedelta(seconds=max(0.0, float(grace_seconds)))) + boundary = boundary.floor("min").tz_localize(None) + stable: set[pd.Timestamp] = set() + for value in validated_frame["timestamp"]: + timestamp = pd.Timestamp(_as_aware_ist(value)).tz_localize(None) + if timestamp < boundary: + stable.add(timestamp) + return frozenset(stable) + + @dataclass(frozen=True) class MarketDataHealthSnapshot: """Immutable worker-facing view of the current feed safety state. diff --git a/Dependencies/session_state.py b/Dependencies/session_state.py index 79df474..f28137a 100644 --- a/Dependencies/session_state.py +++ b/Dependencies/session_state.py @@ -94,6 +94,20 @@ # lines in one session (2026-08-11) and buried the 13 that actually mattered. SLOW_MARKS_WRITE_WARNING_SECONDS = 2.0 +# How far the marks file may lag the durable document before its open positions +# stop being trustworthy. This exists because of 2026-08-12: the snapshot loop +# silently stopped at 12:30 while trading continued to 15:10, so the marks on +# disk described a 2h40m-old book -- 11 positions where 23 were actually open. +# Nothing warned, and nothing would have stopped a resume from restoring that +# stale set as if it were current, which is precisely the "invent exposure" +# failure ADR-0012 exists to prevent. +MAX_RESUMABLE_MARKS_AGE_SECONDS = 300.0 + +# How long the supervisor may go without a successful marks write before the +# store says so. Deliberately a multiple of the snapshot interval rather than a +# fixed number, so a slow disk that merely delays a write does not cry wolf. +MARKS_STALL_INTERVALS = 4 + # Per-strategy keys that live in the MARKS file rather than the durable one. # All of them are refreshed wholesale by every snapshot, so losing the newest # 30 seconds of them in a crash is the documented trade-off (ADR-0012); none is @@ -297,6 +311,24 @@ def __init__( # permissions, missing drive) logs one error instead of one per trade. self._write_failure_logged = False + # Liveness bookkeeping for the supervisor heartbeat and the stall guard. + # `_marks_writes` counts SUCCESSFUL marks publishes; the two timestamps + # are monotonic (immune to a clock step) and wall-clock (readable in a + # log line) views of the newest one. They start at construction so a loop + # that never runs at all still reports a growing age rather than None. + self._marks_writes = 0 + self._last_marks_write_monotonic = time.monotonic() + self._last_marks_write_at = _now_ist() + self._marks_stall_logged = False + + # Optional off-thread durable writer (see start_durable_writer). While + # `_writer` is None every trade event is persisted synchronously by the + # calling trading thread, which is the original ADR-0012 contract. + self._writer: threading.Thread | None = None + self._writer_stopping = False + self._durable_dirty = False + self._durable_wakeup = threading.Event() + self._state: dict[str, Any] = { "schema_version": SCHEMA_VERSION, "session_date": self.session_date.isoformat(), @@ -422,10 +454,110 @@ def record_trade_event(self, event: Mapping[str, Any]) -> None: # Drop oldest first -- a recovery cares about the newest. del trades[: len(trades) - self.max_trade_records] self._apply_pnl_bearing_event_locked(record) - self._flush_durable_locked() + if self._writer is None: + # Synchronous: the caller does not continue until the event + # is on the platter. Costs the caller the whole fsync. + self._flush_durable_locked() + return + # Asynchronous: the in-memory state is already updated, so the + # writer thread will persist THIS event and any that arrive + # while it works, in one coalesced write. + self._durable_dirty = True + self._durable_wakeup.set() except Exception: # noqa: BLE001 - reporting must never break trading self._log_write_failure("record trade event") + # ------------------------------------------------------------------ + # Optional off-thread durable writer + # ------------------------------------------------------------------ + # Why this exists: the synchronous write blocks the TRADING thread that + # published the event. On the operator's hardware that is a median 0.65s + # and up to 2.3s per event, and the stalls are not merely latency -- on + # 2026-08-11, 86% of the websocket feed disconnects (25 of 29) landed within + # 20s of a session-state write stall, with `keepalive ping timeout` the + # dominant error. A blocked event loop drops the market feed. + # + # What it costs: a hard kill can lose events that are queued but not yet + # written, where the synchronous path loses only the one in flight. Note + # the data does NOT land later -- the write takes the same time either way, + # so it reaches the platter at the same wall-clock moment; the difference is + # that the trading thread is not frozen meanwhile. The `EXIT` log line is + # also emitted BEFORE publish_trade_event, so a lost queued event still has + # a log record, which is the same source the EOD Sheet parses. + # + # Coalescing makes the exposure smaller than it first appears: the document + # is a full rewrite, so a burst of queued events becomes ONE write. That + # reduces total fsyncs as well as moving them off the trading path. + def start_durable_writer(self) -> None: + """Begin persisting trade events on a background thread.""" + with self._lock: + if self._writer is not None: + return + self._writer_stopping = False + self._writer = threading.Thread( + target=self._durable_writer_loop, + name="SessionStateWriter", + daemon=True, + ) + self._writer.start() + self.log.info( + "Session state durable writes are ASYNCHRONOUS: trade events are " + "persisted by a background writer, so a trading thread is never " + "blocked by fsync. A hard kill can lose events queued in the last " + "write cycle; their log lines survive." + ) + + def stop_durable_writer(self, timeout: float = 10.0) -> bool: + """Drain and stop the writer. Returns True when everything was written. + + Called at shutdown BEFORE results are published, so an orderly end of + day is exactly as durable as the synchronous path. + """ + with self._lock: + writer = self._writer + if writer is None: + return True + self._writer_stopping = True + self._durable_wakeup.set() + writer.join(timeout=timeout) + drained = not writer.is_alive() + with self._lock: + self._writer = None + pending = self._durable_dirty + if pending: + # Last resort: persist synchronously on the caller's thread rather + # than leave a recorded trade unwritten. + try: + with self._lock: + self._flush_durable_locked() + self._durable_dirty = False + except Exception: # noqa: BLE001 - shutdown must still complete + self._log_write_failure("drain the durable writer") + return False + return drained + + def _durable_writer_loop(self) -> None: + """Coalesce pending trade events into one atomic write at a time.""" + while True: + self._durable_wakeup.wait(timeout=1.0) + self._durable_wakeup.clear() + try: + with self._lock: + stopping = self._writer_stopping + dirty = self._durable_dirty + if dirty: + # Clear BEFORE writing: events arriving during the write + # re-set the flag and earn their own next cycle, so none + # is silently folded into a write that already started. + self._durable_dirty = False + self._flush_durable_locked() + except Exception: # noqa: BLE001 - the writer must never die quietly + self._log_write_failure("write session state from the writer thread") + with self._lock: + self._durable_dirty = True + if stopping and not dirty: + return + def _apply_pnl_bearing_event_locked(self, record: Mapping[str, Any]) -> None: """Fold a realized-P&L event into that strategy's running totals. @@ -621,6 +753,13 @@ def _flush_marks_locked(self) -> None: elapsed = self._write_document( self.marks_path, self._marks_document_locked(), durable=False ) + # Only a COMPLETED publish counts. The 2026-08-12 freeze produced no + # error and no partial file, so "did a write finish" is the only signal + # that distinguishes a healthy loop from a stopped one. + self._marks_writes += 1 + self._last_marks_write_monotonic = time.monotonic() + self._last_marks_write_at = _now_ist() + self._marks_stall_logged = False if elapsed >= SLOW_MARKS_WRITE_WARNING_SECONDS: self.log.warning( "Session state marks write took %.3fs (path=%s); this delays " @@ -649,6 +788,56 @@ def snapshot(self) -> dict[str, Any]: with self._lock: return json.loads(json.dumps(self._state)) + def health(self) -> dict[str, Any]: + """Liveness of the snapshot loop, for the supervisor heartbeat. + + `marks_stalled` is the question 2026-08-12 could not answer: the loop + stopped writing at 12:30 while trading ran to 15:10, with no exception, + no partial file and no log line. Counting completed publishes and how + long ago the newest one was makes a stopped loop observable within a few + minutes instead of at end-of-day. + """ + with self._lock: + age = time.monotonic() - self._last_marks_write_monotonic + open_positions = sum( + 1 + for entry in self._state.get("strategies", {}).values() + if isinstance(entry, Mapping) and entry.get("open_position") + ) + return { + "marks_writes": self._marks_writes, + "marks_age_seconds": round(age, 1), + "marks_last_write_at": self._last_marks_write_at.isoformat(), + "marks_stalled": age > (self.snapshot_interval_seconds * MARKS_STALL_INTERVALS), + "trades_recorded": len(self._state.get("trades", [])), + "open_positions": open_positions, + "write_failures_logged": self._write_failure_logged, + } + + def warn_if_marks_stalled(self) -> bool: + """Log ONCE per stall episode that the snapshot loop has gone quiet. + + Returns True when a stall is currently detected (whether or not this + call did the logging). Kept here rather than in the runner so the + threshold and the one-shot latch live beside the counters they read. + """ + report = self.health() + if not report["marks_stalled"]: + return False + if not self._marks_stall_logged: + self._marks_stall_logged = True + self.log.error( + "Session state marks have not been written for %.0fs (expected every " + "%.0fs, last at %s, %d writes so far). Open-position recovery for this " + "session is DEGRADED: the marks file describes an older book than the " + "durable trade log. Realized P&L is unaffected.", + report["marks_age_seconds"], + self.snapshot_interval_seconds, + report["marks_last_write_at"], + report["marks_writes"], + ) + return True + def _marks_path_for(path: str | Path) -> Path: """Sibling marks path for a durable state path (``x.json`` -> ``x.marks.json``).""" @@ -714,6 +903,11 @@ def load_session_state(path: str | Path) -> dict[str, Any] | None: ) return state + # How far the marks lag the durable document. The durable file is stamped on + # every trade event, so this is a direct measure of "how much trading + # happened after the last snapshot" -- which on 2026-08-12 was 2h40m. + state["marks_age_seconds"] = _document_lag_seconds(state, marks) + mark_strategies = marks.get("strategies") if not isinstance(mark_strategies, Mapping): return state @@ -729,6 +923,22 @@ def load_session_state(path: str | Path) -> dict[str, Any] | None: return state +def _document_lag_seconds( + state: Mapping[str, Any], marks: Mapping[str, Any] +) -> float | None: + """Seconds by which the marks document trails the durable one. + + ``None`` when either timestamp is missing or unparseable — the caller treats + an unknown lag as untrustworthy rather than as zero. + """ + try: + durable_at = datetime.fromisoformat(str(state.get("updated_at", ""))) + marks_at = datetime.fromisoformat(str(marks.get("updated_at", ""))) + except ValueError: + return None + return round((durable_at - marks_at).total_seconds(), 1) + + def resumable_open_positions( state: Mapping[str, Any] | None, *, @@ -766,6 +976,23 @@ def resumable_open_positions( if bool(state.get("clean_shutdown", False)): return {} + # STALE MARKS ARE NOT A BOOK. The positions live in the best-effort marks + # file; if the snapshot loop stopped while trading continued, that file + # describes an older set of positions than actually existed. Restoring it + # would invent exposure that was closed and omit exposure that was opened -- + # exactly what happened on 2026-08-12, where a 2h40m-stale file held 11 + # positions against 23 genuinely open. Refuse rather than half-restore. + lag = state.get("marks_age_seconds") + if lag is None or float(lag) > MAX_RESUMABLE_MARKS_AGE_SECONDS: + logger.warning( + "Not resuming any position: the marks file lags the durable trade log " + "by %s (limit %.0fs), so its open positions are not a current book. " + "Realized P&L is unaffected; square off manually against the broker.", + "an unknown amount" if lag is None else f"{float(lag):.0f}s", + MAX_RESUMABLE_MARKS_AGE_SECONDS, + ) + return {} + resumable: dict[str, dict[str, Any]] = {} strategies = state.get("strategies") if not isinstance(strategies, Mapping): diff --git a/Nifty Multi Strategy Front Test - Master File.py b/Nifty Multi Strategy Front Test - Master File.py index 3b40326..560412e 100644 --- a/Nifty Multi Strategy Front Test - Master File.py +++ b/Nifty Multi Strategy Front Test - Master File.py @@ -253,6 +253,7 @@ import time import uuid import warnings +from collections.abc import Iterable from dataclasses import dataclass from datetime import date, datetime, timedelta from datetime import time as dt_time @@ -289,6 +290,7 @@ MarketDataValidationError, complete_minute_bucket_mask, newest_completed_minute_timestamp, + stable_official_minutes, validate_ohlc_frame, ) from Dependencies.next_open_entry import PendingNextOpenEntry @@ -652,6 +654,26 @@ def _scaled_float(prefix: str, name: str, default: float) -> float: # and the runner already reconciles against it; a JSON file that disagrees with # the account is worse than no file at all. See `_resume_open_positions`. SESSION_STATE_RESUME_ENABLED = _env_bool("SESSION_STATE_RESUME_ENABLED", False) +# How often the supervisor logs a liveness line. MainThread is otherwise silent +# through a healthy session, so when the snapshot loop stopped on 2026-08-12 its +# silence carried no information and the freeze was only found hours later from +# a file mtime. Five minutes is ~75 lines a session -- cheap enough to keep on +# always, frequent enough to localise a stall to a few minutes. +SESSION_STATE_HEARTBEAT_SECONDS = _env_float("SESSION_STATE_HEARTBEAT_SECONDS", 300.0) +# Persist trade events from a background writer instead of on the trading thread +# that published them. OFF by default because it trades a small amount of the +# durability ADR-0012 exists to provide: a hard kill can lose events queued in +# the last write cycle, where the synchronous path loses only the one in flight. +# +# Turn it ON if disk stalls are costing you the market feed. On this operator's +# hardware they are: on 2026-08-11, 86% of websocket disconnects (25 of 29) +# landed within 20 seconds of a session-state write stall, with `keepalive ping +# timeout` the dominant error -- a blocked event loop drops the feed. The data +# still reaches the platter at the same moment either way (the write takes the +# same time); what changes is that no trading thread waits for it. Lost queued +# events also still have their `EXIT` log line, which is what the EOD Sheet +# parses, so the fallback is the same one used before this module existed. +SESSION_STATE_ASYNC_WRITES = _env_bool("SESSION_STATE_ASYNC_WRITES", False) # Telegram trade-notification settings. See Dependencies/.env for the one-time # bot/channel setup. When disabled (or token/chat blank) the notifier thread is @@ -1904,8 +1926,10 @@ class MarketSnapshot: - `source_candle_ts` : timestamp of the latest candle in `frame`. - `candle_signature` : lightweight fingerprint of the latest row's state. - `fetched_at` : wall-clock time when the fetch completed. - - `official_candle_ts`: newest row supplied by the REST/official source, or - ``None`` when the publisher has not proved any official coverage. + - `official_completed_minutes`: exact immutable REST minute stamps proven + final for this generation. + - `official_candle_ts`: compatibility maximum derived from that exact set, + or ``None`` when the publisher has not proved any official coverage. Why we keep both `source_candle_ts` and `candle_signature`: - During a live 1-minute candle, the timestamp does not change but the @@ -1922,8 +1946,12 @@ class MarketSnapshot: candle_signature: tuple | None fetched_at: datetime # Websocket frames can contain a mix of official REST history and newer - # tick-built rows. This watermark tells a five-minute consumer exactly how - # far the official portion reaches without changing the canonical OHLC data. + # tick-built rows. CPR needs the exact REST minute identities, not merely a + # maximum: a missing 09:57 must block the 09:55 five-minute bucket even when + # official data already reaches 09:59. + official_completed_minutes: frozenset[pd.Timestamp] = frozenset() + # Compatibility view for existing non-CPR consumers. This is always derived + # from ``official_completed_minutes`` by SharedMarketDataStore.update(). official_candle_ts: pd.Timestamp | None = None @@ -2227,6 +2255,7 @@ def update( timeframe: str, frame: pd.DataFrame, *, + official_completed_minutes: Iterable[pd.Timestamp | datetime] | None = None, official_candle_ts: pd.Timestamp | datetime | None = None, ) -> MarketSnapshot: """ @@ -2235,13 +2264,29 @@ def update( The new snapshot is built BEFORE the lock is taken; only the swap happens under lock so the critical section stays small. - ``official_candle_ts`` is optional because a pure tick publisher has no - official coverage to claim. REST-backed publishers pass their newest - source timestamp so conservative consumers can wait for official data. + ``official_completed_minutes`` is optional because a pure tick publisher + has no official coverage to claim. REST-backed publishers pass every + proven-final minute in the same update as the matching frame. The legacy + ``official_candle_ts`` argument remains a compatibility input for older + callers; it is converted to a one-element exact collection. """ validated = validate_ohlc_frame(frame) source_candle_ts = pd.to_datetime(validated.iloc[-1]["timestamp"]) candle_signature = build_last_row_signature(validated) + if official_completed_minutes is None: + raw_official_minutes = () if official_candle_ts is None else (official_candle_ts,) + else: + raw_official_minutes = official_completed_minutes + normalized_official_values: set[pd.Timestamp] = set() + for value in raw_official_minutes: + timestamp = pd.Timestamp(value) + if pd.isna(timestamp): + continue + if timestamp.tzinfo is not None: + timestamp = timestamp.tz_convert(ZoneInfo("Asia/Kolkata")).tz_localize(None) + normalized_official_values.add(timestamp) + normalized_official_minutes = frozenset(normalized_official_values) + derived_official_candle_ts = max(normalized_official_minutes, default=None) snapshot = MarketSnapshot( timeframe=str(timeframe), @@ -2249,11 +2294,8 @@ def update( source_candle_ts=source_candle_ts, candle_signature=candle_signature, fetched_at=datetime.now(ZoneInfo("Asia/Kolkata")), - official_candle_ts=( - None - if official_candle_ts is None - else pd.Timestamp(official_candle_ts) - ), + official_completed_minutes=normalized_official_minutes, + official_candle_ts=derived_official_candle_ts, ) with self._lock: self._snapshots[str(timeframe)] = snapshot @@ -2277,6 +2319,7 @@ def get(self, timeframe: str) -> MarketSnapshot | None: source_candle_ts=snapshot.source_candle_ts, candle_signature=snapshot.candle_signature, fetched_at=snapshot.fetched_at, + official_completed_minutes=snapshot.official_completed_minutes, official_candle_ts=snapshot.official_candle_ts, ) @@ -3817,14 +3860,20 @@ def run(self) -> None: if self.stop_event.is_set(): break try: + # The request-start time, not the response arrival time, + # decides whether Dhan's final row was still forming. + request_started_at = _ist_now() frame = self.fetch_ohlc(timeframe) - # This producer's entire frame came directly from REST, so - # its newest timestamp is also the official-data watermark. - # Websocket mode publishes the same metadata after true-up. + validated = validate_ohlc_frame(frame) + completed_minutes = stable_official_minutes( + validated, + request_started_at=request_started_at, + grace_seconds=WS_TRUEUP_DELAY_SECONDS, + ) snapshot = self.store.update( timeframe, - frame, - official_candle_ts=pd.Timestamp(frame["timestamp"].max()), + validated, + official_completed_minutes=completed_minutes, ) if self.last_logged_candle_ts.get(timeframe) != snapshot.source_candle_ts: self.last_logged_candle_ts[timeframe] = snapshot.source_candle_ts @@ -3880,9 +3929,11 @@ class WebSocketMarketDataFetcher(threading.Thread): the full desired instrument set by construction. Bar semantics: the published frame is always - ``merge_official_and_tick_frames(official REST history, tick bars)`` -- - official candles win for completed minutes, the forming minute is always - tick-built, and every publish still goes through `store.update()` and its + ``merge_official_and_tick_frames(official REST history, tick bars)``. + Official REST candles win only for minutes proved final at the REST + request's start. The forming minute and a newly closed minute still inside + the grace period remain tick-built until a later true-up proves them final. + Every publish still goes through `store.update()` and its `validate_ohlc_frame` net. Consumers are untouched: same store, same snapshot shape, same health gates. """ @@ -3921,10 +3972,10 @@ def __init__( # Latest REST history: warmup seed, then refreshed by every true-up. # Supervisor-owned; the pump never touches it. self.official_frame: pd.DataFrame = pd.DataFrame() - # The frame may later be merged with newer tick-built rows. Keep the - # newest timestamp that came from REST as a separate watermark so a - # strategy can prove its completed bucket has been officially trued up. - self._official_candle_ts: pd.Timestamp | None = None + # The frame may later be merged with newer tick-built rows. Keep the + # exact REST minutes proved final for this generation so CPR can reject + # a five-minute bucket with an intermediate official-data hole. + self._official_completed_minutes: frozenset[pd.Timestamp] = frozenset() # Connection state shared between pump and supervisor. self._feed_lock = threading.Lock() @@ -4015,17 +4066,28 @@ def _warmup_official_history(self) -> bool: index_key = (NIFTY_INDEX_EXCHANGE_SEGMENT, NIFTY_INDEX_SECURITY_ID) while not self.stop_event.is_set(): try: + request_started_at = _ist_now() frame = self.broker.fetch_index_1m_ohlc( security_id=NIFTY_INDEX_SECURITY_ID, exchange_segment=NIFTY_INDEX_EXCHANGE_SEGMENT, instrument_type=NIFTY_INDEX_INSTRUMENT_TYPE, ) - self.official_frame = frame - self._official_candle_ts = pd.Timestamp(frame["timestamp"].max()) + validated = validate_ohlc_frame(frame) + completed_minutes = stable_official_minutes( + validated, + request_started_at=request_started_at, + grace_seconds=WS_TRUEUP_DELAY_SECONDS, + ) + # REST may include the live or grace-period minute. Leave those + # rows entirely tick-owned until a later request proves them final. + self.official_frame = validated.loc[ + validated["timestamp"].isin(completed_minutes) + ].reset_index(drop=True) + self._official_completed_minutes = completed_minutes self.store.update( "1", - frame, - official_candle_ts=self._official_candle_ts, + self.official_frame, + official_completed_minutes=completed_minutes, ) self.log.info("Warmup history loaded | Rows=%s", len(frame)) return True @@ -4156,7 +4218,7 @@ def _publish_frame_locked(self, force: bool) -> None: snapshot = self.store.update( "1", frame, - official_candle_ts=self._official_candle_ts, + official_completed_minutes=self._official_completed_minutes, ) self._ohlc_ok = True if self.last_logged_candle_ts != snapshot.source_candle_ts: @@ -4207,13 +4269,23 @@ def _maybe_run_true_up(self, now_ist: datetime | None = None) -> None: def _run_true_up(self, reason: str, now_ist: datetime | None = None) -> None: """ - Replace completed candles with Dhan's official ones (tick bars stay - for anything REST does not cover, most importantly the forming - minute). A REST failure keeps serving tick bars and retries on the - next minute -- the tick feed remains the live source of truth. + Replace only REST minutes proved final at the request-start boundary. + + Tick bars remain in charge of the forming minute and any newer row that + is still inside the grace period. An older hole in stable REST history + stays fail-closed until a later REST response fills it; we do not keep a + provisional tick candle as if it were official. After a REST failure, + the next active minute with fresh ticks creates another true-up chance. """ if now_ist is None: now_ist = datetime.now(ZoneInfo("Asia/Kolkata")).replace(tzinfo=None) + # Record the actual request boundary before any blocking HTTP work. + # Test callers pass ``now_ist`` as that deterministic request-start clock. + request_started_at = ( + now_ist.replace(tzinfo=IST_TIMEZONE) + if now_ist is not None and now_ist.tzinfo is None + else now_ist or _ist_now() + ) try: official = self.broker.fetch_index_1m_ohlc( security_id=NIFTY_INDEX_SECURITY_ID, @@ -4228,25 +4300,34 @@ def _run_true_up(self, reason: str, now_ist: datetime | None = None) -> None: if official is None or official.empty: self.log.warning("True-up (%s) returned no official candles.", reason) return + validated = validate_ohlc_frame(official) + completed_minutes = stable_official_minutes( + validated, + request_started_at=request_started_at, + grace_seconds=WS_TRUEUP_DELAY_SECONDS, + ) + stable_official = validated.loc[ + validated["timestamp"].isin(completed_minutes) + ].reset_index(drop=True) stats = divergence_stats( - official, + stable_official, self.aggregator.tick_bars_frame(), forming_minute=pd.Timestamp(now_ist).floor("min"), ) - newest_official = pd.Timestamp(official["timestamp"].max()) # Update the official frame, its watermark, and the published merged # snapshot under one lock. Without this atomic boundary, the supervisor # could publish new official OHLC with an older watermark (or vice versa) # and make a waiting CPR worker observe a mixed generation. with self._publish_lock: - self.official_frame = official - self._official_candle_ts = newest_official + self.official_frame = stable_official + self._official_completed_minutes = completed_minutes self._publish_frame_locked(force=True) - # Drop every tick bar the official history now covers. The merge would - # ignore them anyway (official wins), but keeping them makes the NEXT - # divergence report re-count this cycle's mismatches forever -- the - # stats above must describe only the minutes trued-up right now. - self.aggregator.prune_older_than(newest_official + pd.Timedelta(minutes=1)) + newest_official = max(completed_minutes, default=None) + if newest_official is not None: + # Prune only through the newest stable official minute. A REST row + # still inside the grace period must remain tick-owned and available + # for the later, final true-up. + self.aggregator.prune_older_than(newest_official + pd.Timedelta(minutes=1)) log_fn = ( self.log.warning if stats.mismatched and stats.max_abs_delta > self.TRUEUP_DIVERGENCE_WARN_POINTS @@ -4255,7 +4336,7 @@ def _run_true_up(self, reason: str, now_ist: datetime | None = None) -> None: log_fn( "True-up (%s) | OfficialRows=%s | Overlap=%s | Mismatched=%s | MaxAbsDelta=%.2f", reason, - len(official), + len(stable_official), stats.overlapping, stats.mismatched, stats.max_abs_delta, @@ -9674,25 +9755,92 @@ def _official_snapshot_covers_completed_bar( snapshot: MarketSnapshot, strategy_frame: pd.DataFrame, ) -> bool: - """Return true only when REST covers the bucket's final source minute. + """Return true only when REST covers every exact source minute. Five-minute candles are start-stamped. A 09:55 candle therefore needs - the official one-minute source to cover 09:59 before Codex may freeze - it. This condition-based gate naturally tolerates a slow Dhan response: - the normal worker poll simply checks the next atomic snapshot instead - of guessing how many seconds the REST true-up will take. + every official one-minute source from 09:55 through 09:59 before Codex + may freeze it. A maximum timestamp alone is unsafe because a REST hole + (for example missing 09:57) would otherwise authorize invented OHLC. + This exact-set gate naturally tolerates a slow Dhan response: the normal + worker poll simply checks the next atomic snapshot instead of guessing + how many seconds the REST true-up will take. """ if strategy_frame.empty or "timestamp" not in strategy_frame.columns: return False - official_timestamp = self._naive_ist_timestamp(snapshot.official_candle_ts) bucket_start = self._naive_ist_timestamp( strategy_frame.iloc[-1]["timestamp"] ) - if official_timestamp is None or bucket_start is None: + if bucket_start is None: return False - bucket_final_minute = bucket_start + pd.Timedelta(minutes=4) - return official_timestamp >= bucket_final_minute + required_minutes = frozenset( + bucket_start + pd.Timedelta(minutes=offset) for offset in range(5) + ) + official_minutes = frozenset( + timestamp + for value in snapshot.official_completed_minutes + if (timestamp := self._naive_ist_timestamp(value)) is not None + ) + return required_minutes.issubset(official_minutes) + + def _completed_bar_audit_metadata( + self, + snapshot: MarketSnapshot | None, + strategy_frame: pd.DataFrame, + frozen_signature: str, + ) -> dict[str, object]: + """Capture exact official-minute coverage once for both audit records. + + The worker already proved this coverage before inference in ``run``. + Keeping the same small, JSON-ready mapping through pre- and post-action + logging makes the two rows comparable without re-reading a market store + that may have advanced while Codex or execution was running. Direct + unit/diagnostic callers may not have a snapshot; they receive an empty, + explicitly false coverage record rather than invented evidence. + + ``required_official_minutes`` lists the five one-minute source rows + needed to build this bucket. ``present_official_minutes`` is only the + intersection of those five rows with the frozen official set -- not all + REST history in the store. ``official_coverage`` is true only when that + intersection contains all five required rows. + """ + + bucket_start = ( + None + if strategy_frame.empty or "timestamp" not in strategy_frame.columns + else self._naive_ist_timestamp(strategy_frame.iloc[-1]["timestamp"]) + ) + required_minutes = frozenset( + bucket_start + pd.Timedelta(minutes=offset) + for offset in range(5) + ) if bucket_start is not None else frozenset() + official_minutes = frozenset( + timestamp + for value in ( + () if snapshot is None else snapshot.official_completed_minutes + ) + if (timestamp := self._naive_ist_timestamp(value)) is not None + ) + + def ist_iso(timestamp: pd.Timestamp) -> str: + """Represent internal naive-IST timestamps unambiguously in JSONL.""" + + return timestamp.tz_localize(IST_TIMEZONE).isoformat() + + return { + "bar_timestamp": None if bucket_start is None else ist_iso(bucket_start), + "frozen_signature": frozen_signature, + "required_official_minutes": [ + ist_iso(timestamp) for timestamp in sorted(required_minutes) + ], + "present_official_minutes": [ + ist_iso(timestamp) + for timestamp in sorted(required_minutes & official_minutes) + ], + "official_coverage": bool(required_minutes) and required_minutes.issubset( + official_minutes + ), + } def _position_state_payload(self) -> dict[str, object]: """Expose allowlisted premise/risk facts, never execution capabilities. @@ -10372,6 +10520,7 @@ def _write_final_execution( frozen_context: dict[str, object], outcome, execution: dict[str, object], + bar_metadata: dict[str, object], ) -> None: """Best-effort append actual post-action provenance to the decision log. @@ -10392,6 +10541,8 @@ def _write_final_execution( token_usage=outcome.token_usage, tool_evidence=self._tool_log_payload(outcome.tool_evidence), execution=execution, + audit_stage="POST_ACTION", + bar_metadata=bar_metadata, ) except Exception as exc: # noqa: BLE001 - the pre-action audit still exists self.log.error( @@ -10423,7 +10574,12 @@ def _post_inference_exposure_block_reason(self) -> str: return "entry_cutoff" return "" - def process_strategy_frame(self, strategy_frame: pd.DataFrame) -> None: + def process_strategy_frame( + self, + strategy_frame: pd.DataFrame, + *, + audit_metadata: dict[str, object] | None = None, + ) -> None: """Evaluate one completed bucket through mechanics, Codex, and host gates. Flat workers skip new turns after 15:00; open workers continue for @@ -10448,6 +10604,13 @@ def process_strategy_frame(self, strategy_frame: pd.DataFrame) -> None: bar_signature = self._completed_bar_signature(strategy_frame) if not bar_signature: return + if audit_metadata is None: + # Tests and direct diagnostics can invoke this method outside the + # normal ``run`` loop. Capture a conservative local snapshot once + # so every row still has explicit, non-invented coverage facts. + audit_metadata = self._completed_bar_audit_metadata( + self.store.get(self.timeframe), strategy_frame, bar_signature + ) frozen_context = self._latest_frozen_context() if self.pos.active and self._manage_completed_bar(frozen_context): return @@ -10496,6 +10659,8 @@ def process_strategy_frame(self, strategy_frame: pd.DataFrame) -> None: else "AUDITED_BEFORE_EXECUTION" ), }, + audit_stage="PRE_ACTION", + bar_metadata=audit_metadata, ) except Exception as exc: # noqa: BLE001 - entries fail closed; exits continue audit_ok = False @@ -10539,6 +10704,7 @@ def process_strategy_frame(self, strategy_frame: pd.DataFrame) -> None: "submitted": True, "status": "EXIT_CONFIRMED" if not self.pos.active else "EXIT_UNCONFIRMED", }, + audit_metadata, ) return if outcome.action == "SCALE_IN" and audit_ok: @@ -10553,6 +10719,7 @@ def process_strategy_frame(self, strategy_frame: pd.DataFrame) -> None: "status": "SCALE_IN_BLOCKED", "blocked_reason": blocked_reason, }, + audit_metadata, ) # A closed exposure gate prevents the add immediately. The # normal safety pass also performs any associated lifecycle, @@ -10580,6 +10747,7 @@ def process_strategy_frame(self, strategy_frame: pd.DataFrame) -> None: else "SCALE_IN_UNCONFIRMED" ), }, + audit_metadata, ) return if not audit_ok or outcome.action not in {"ENTER_LONG", "ENTER_SHORT"}: @@ -10598,6 +10766,7 @@ def process_strategy_frame(self, strategy_frame: pd.DataFrame) -> None: "status": "ENTRY_BLOCKED", "blocked_reason": blocked_reason, }, + audit_metadata, ) return direction = "LONG" if outcome.action == "ENTER_LONG" else "SHORT" @@ -10624,6 +10793,7 @@ def process_strategy_frame(self, strategy_frame: pd.DataFrame) -> None: else "ENTRY_BLOCKED" ), }, + audit_metadata, ) def run(self) -> None: @@ -10631,10 +10801,10 @@ def run(self) -> None: The worker starts decisions at 09:30 but safety runs even before then. Shared one-minute data is resampled with the current IST clock so forming - websocket minutes are excluded. A clock-complete bucket then waits until - the atomic REST watermark covers its fifth source minute. Any single poll - failure is logged and retried; it does not terminate future hard-stop or - square-off handling. + websocket minutes are excluded. A clock-complete bucket then waits for + all five exact official REST source minutes; an intermediate REST hole + keeps the bucket blocked. Any single poll failure is logged and retried; + it does not terminate future hard-stop or square-off handling. """ self.log.info("Starting %s strategy worker.", self.strategy_name) @@ -10669,14 +10839,19 @@ def run(self) -> None: if waiting_identity != self._waiting_for_official_bar_identity: self._waiting_for_official_bar_identity = waiting_identity self.log.info( - "Completed CPR bucket %s is waiting for its final " - "one-minute REST true-up.", + "Completed CPR bucket %s is waiting for all five " + "official source minutes (including any REST hole).", waiting_identity, ) self.wait_for_next_poll() continue self._waiting_for_official_bar_identity = None - self.process_strategy_frame(completed) + audit_metadata = self._completed_bar_audit_metadata( + snapshot, + completed, + self._completed_bar_signature(completed), + ) + self.process_strategy_frame(completed, audit_metadata=audit_metadata) except Exception as exc: # noqa: BLE001 - one turn must not kill safety self.log.exception("CPR AI worker poll failed: %s", exc) self.wait_for_next_poll() @@ -17335,6 +17510,60 @@ def _paper_position_from_record(record: dict) -> PaperPosition: ) +def _emit_supervisor_heartbeat( + session_state: SessionStateStore, + started_workers: list[BasePaperStrategyWorker], + last_heartbeat_at: float | None, +) -> float: + """Log one supervisor liveness line per interval; return the new stamp. + + Why this exists: on 2026-08-12 the snapshot loop stopped writing marks at + 12:30 while workers traded on until 15:10. There was no exception, no + partial file and no log line, so the only evidence was a file mtime found + hours later -- and it was impossible to tell afterwards whether MainThread + had blocked, the loop had exited, or writes were failing silently. + + MainThread otherwise logs nothing during a healthy session, so its silence + carried no information. A periodic line fixes that in both directions: its + presence shows the loop is turning, and the gap between the last heartbeat + and the crash localises where it stopped. + + ``last_heartbeat_at`` is ``None`` until the first line is emitted, and that + sentinel matters: `time.monotonic()` has an arbitrary origin, so a plain 0.0 + means "long ago" on a Windows box whose counter is machine uptime and "just + now" in a freshly booted Linux container. CI caught exactly that -- the first + heartbeat fired locally and was suppressed for five minutes on the runner. + + Never raises: a diagnostic that can take the supervisor down is worse than + no diagnostic at all. + """ + now = time.monotonic() + if ( + last_heartbeat_at is not None + and (now - last_heartbeat_at) < SESSION_STATE_HEARTBEAT_SECONDS + ): + return last_heartbeat_at + try: + report = session_state.health() + # An actual stall is an ERROR in its own right, logged once per episode + # by the store rather than repeated on every heartbeat. + session_state.warn_if_marks_stalled() + logger.info( + "Supervisor heartbeat | workers_alive=%d/%d | marks_writes=%d | " + "marks_age=%.0fs | open_positions=%d | trades_recorded=%d%s", + sum(1 for worker in started_workers if worker.is_alive()), + len(started_workers), + report["marks_writes"], + report["marks_age_seconds"], + report["open_positions"], + report["trades_recorded"], + " | PERSISTENCE DEGRADED" if report["write_failures_logged"] else "", + ) + except Exception: # noqa: BLE001 - a heartbeat must never stop supervision + logger.exception("Supervisor heartbeat failed; supervision continues.") + return now + + def _start_and_supervise_runtime_threads( fetcher: CentralMarketDataFetcher, telegram_worker: TelegramMessageWorker | None, @@ -17352,6 +17581,12 @@ def _start_and_supervise_runtime_threads( started_workers: list[BasePaperStrategyWorker] = [] shutdown_reason = "" + # Monotonic so an NTP step cannot silence or spam the heartbeat. None means + # "never emitted", so the first supervised tick always logs one and proves + # the loop was entered at all -- the 2026-08-12 freeze left no such evidence. + # It must NOT be 0.0: monotonic()'s origin is arbitrary, so 0.0 reads as + # "long ago" on Windows and "just now" in a fresh Linux container. + last_heartbeat_at: float | None = None try: fetcher.start() if telegram_worker is not None: @@ -17371,6 +17606,9 @@ def _start_and_supervise_runtime_threads( # immediately by publish_trade_event. if session_state is not None: session_state.update_worker_snapshot(_session_state_snapshots(started_workers)) + last_heartbeat_at = _emit_supervisor_heartbeat( + session_state, started_workers, last_heartbeat_at + ) return True except KeyboardInterrupt: shutdown_reason = "KEYBOARD_INTERRUPT" @@ -17976,9 +18214,15 @@ def main() -> None: # One immediate write so the file exists (and is stamped with this # session's date) even if the process dies before the first trade. session_state.update_worker_snapshot(_session_state_snapshots(workers), force=True) + if SESSION_STATE_ASYNC_WRITES: + # Started AFTER the first snapshot so the file exists before any + # background writing begins. + session_state.start_durable_writer() logger.info( - "Session state persistence ENABLED -> %s (snapshot every %.0fs, resume=%s).", - SESSION_STATE_FILE, SESSION_STATE_SNAPSHOT_SECONDS, SESSION_STATE_RESUME_ENABLED, + "Session state persistence ENABLED -> %s (snapshot every %.0fs, " + "resume=%s, async_writes=%s).", + SESSION_STATE_FILE, SESSION_STATE_SNAPSHOT_SECONDS, + SESSION_STATE_RESUME_ENABLED, SESSION_STATE_ASYNC_WRITES, ) except Exception: # noqa: BLE001 - reporting must never stop a session session_state = None @@ -18031,6 +18275,15 @@ def main() -> None: # Sheet publication is a separate flag: a Ctrl+C shutdown or Google outage # can be locally clean while its figures still need export/reconciliation. if session_state is not None: + # Drain the background writer FIRST so an orderly end of day is exactly + # as durable as the synchronous path: every recorded trade must be on + # disk before the clean-shutdown flag claims the session finished well. + if not session_state.stop_durable_writer(): + logger.error( + "Session state writer did not drain cleanly; the durable file may be " + "missing the last trade event(s). Their log lines remain, and the " + "EOD Sheet parses those." + ) session_state.update_worker_snapshot(_session_state_snapshots(workers), force=True) session_state.mark_clean_shutdown( results_published=finalization.results_published, diff --git a/Signal Generators/CPR AI Agent/README.md b/Signal Generators/CPR AI Agent/README.md index bc12487..7a05954 100644 --- a/Signal Generators/CPR AI Agent/README.md +++ b/Signal Generators/CPR AI Agent/README.md @@ -29,11 +29,13 @@ deep-copy views of the same frozen completed-bar context: The snapshot contains no order surface, account, credential, broker, venue, or execution object. The immediate turn request repeats the four exact tool names -in addition to the developer prompt. If the first isolated turn still omits a -tool, or one tool reports failure, the host permits one retry against the same -immutable snapshot using only the time left in the original SDK deadline. A -second incomplete result, a duplicate/unapproved tool, or any unexpected agent -action invalidates the turn and produces `HOLD`. +in addition to the developer prompt. A repair is considered only when a missing +or failed required tool is the sole remaining defect: the first response must +also still match the current bar, strict schema, configured model/prompt, and +deterministic host policy. The repair uses the same immutable snapshot and only +the time left in the original CPR turn wall-clock deadline, including isolated +child-process overhead. A second incomplete result, a duplicate/unapproved +tool, or any unexpected agent action invalidates the turn and produces `HOLD`. ## Decision contract and host gates @@ -79,13 +81,16 @@ host then enforces the selected framework: each newly completed five-minute candle. A start-stamped one-minute candle is not complete until the next minute begins, and all five exact minute slots must exist once. -- In websocket mode, clock completeness alone is not enough. Every shared OHLC - snapshot carries an atomic `official_candle_ts` watermark. A 09:55 five-minute - bucket waits until the REST source covers its final 09:59 one-minute candle, - even when Dhan's true-up takes longer than its normal five-second delay. This - is a condition check rather than a hard-coded sleep. A later official revision - can still invalidate an in-flight result, but it cannot create a second model - call for an already-consumed bucket. +- In websocket mode, clock completeness alone is not enough. The REST producer + records only a stable generation: a source minute is eligible exactly when + its timestamp is strictly before `floor(request_started_at - true_up_delay)`. + Every shared OHLC snapshot atomically carries that exact immutable set as + `official_completed_minutes`. A 09:55 five-minute bucket needs all five exact + source minutes, 09:55 through 09:59; the final watermark alone is insufficient + because an intermediate REST hole must still block inference. This is a + condition check rather than a hard-coded sleep. A later official revision can + still invalidate an in-flight result, but it cannot create a second model call + for an already-consumed bucket. - At 15:00 IST new entries and adds stop; management and exits continue. - At 15:15 IST the host square-off closes exposure and stops the worker. @@ -144,13 +149,29 @@ missing row with a warning; the labels remain separate from legacy CPR workers. ## Decision audit With `CPR_AI_DECISION_LOGGING_ENABLED=true`, the host appends sanitized JSONL to -`Backtest Outputs/cpr_ai_decisions.jsonl` by default. Each row records the frozen -context, proposal, accepted regime, validation code/reason, authoritative host -geometry, execution outcome, latency, inference-attempt count, aggregate token -usage across a retry, and final tool-call evidence. -Credential-like mapping fields are removed recursively before serialization. -Logging never makes a proposal executable, and an enabled log must succeed -before an entry or add may be submitted. +`Backtest Outputs/cpr_ai_decisions.jsonl` by default. Each row has an IST +`recorded_at` timestamp and an `audit_stage`: `PRE_ACTION` is the host record +before an entry/add may increase exposure, and `POST_ACTION` records the actual +submission/confirmation result afterward. Direct diagnostic callers retain the +safe `DIRECT` stage. + +The `bar` object records the start timestamp, frozen signature, the one current +signature captured when the host finalizes validation or a terminal fail-closed +outcome, all five required official minute stamps, the required stamps present +in the inference snapshot, and the resulting exact coverage boolean. + +Each `attempt_evidence` item records only its request kind (`normal` or the fixed +`tool_repair`), typed evidence result, safe tool name/status records, and token +usage. A provisional timeout marker can appear when a turn was selected but no +child evidence returned before the shared deadline; empty tool/usage fields do +not claim that a launched child consumed zero tokens. Any terminal failure +remains a fail-closed HOLD. + +Credential-like mapping fields are removed recursively before serialization, and +the logger deliberately omits model reasoning/final responses, auth data, local +paths, broker/order/venue details, symbols, quantities, and SDK error text. +Logging never makes a proposal executable, and an enabled log must succeed before +an entry or add may be submitted. ## Zero-order smoke commands diff --git a/Signal Generators/CPR AI Agent/cpr_ai_agent.py b/Signal Generators/CPR AI Agent/cpr_ai_agent.py index 57109a2..7b4aac4 100644 --- a/Signal Generators/CPR AI Agent/cpr_ai_agent.py +++ b/Signal Generators/CPR AI Agent/cpr_ai_agent.py @@ -19,6 +19,7 @@ from collections.abc import Callable, Mapping from concurrent.futures import ThreadPoolExecutor, TimeoutError from dataclasses import dataclass, field +from enum import StrEnum from time import monotonic from typing import Any @@ -30,6 +31,18 @@ _RETRIABLE_TOOL_EVIDENCE_CODES = frozenset({"missing_tool_call", "failed_tool_call"}) +class CPRTurnRequestKind(StrEnum): + """Name the only two fixed turn requests allowed across the child boundary. + + The parent controls this enum. In particular, no model response or caller + string can become a follow-up instruction, which keeps the corrective turn + limited to re-reading the four already-frozen facts. + """ + + NORMAL = "normal" + TOOL_REPAIR = "tool_repair" + + @dataclass(frozen=True) class CPRToolCallRecord: """One SDK-observed MCP read, retained only to prove tool coverage. @@ -57,6 +70,25 @@ class CPRAgentRunResult: unexpected_actions: tuple[str, ...] = () +@dataclass(frozen=True) +class CPRAttemptEvidence: + """Auditable record for one isolated turn slot within a bar deadline. + + Most records describe a child turn that returned. A record can instead be + a conservative pre-launch/timeout marker when the shared deadline expires + before usable child evidence comes back. Empty tool records and token usage + then mean "nothing was returned and proved," not "the child proved it used + zero tokens." The audit never keeps chain-of-thought, returned MCP payloads, + authentication data, or any execution capability. + """ + + attempt_number: int + request_kind: CPRTurnRequestKind + evidence_code: str | None + tool_records: tuple[CPRToolCallRecord, ...] + token_usage: dict[str, int] + + @dataclass class CPRAgentOutcome: """Host-owned outcome separating advice from executable geometry. @@ -83,6 +115,12 @@ class CPRAgentOutcome: token_usage: dict[str, int] = field(default_factory=dict) tool_evidence: tuple[CPRToolCallRecord, ...] = () inference_attempts: int = 1 + attempt_evidence: tuple[CPRAttemptEvidence, ...] = () + # This is the single current-signature sample captured when the host + # finalizes either normal validation or a terminal fail-closed outcome. + # Keeping it on the outcome prevents a later audit writer from accidentally + # observing a newer shared-market generation. + validation_current_signature: str | None = None def _hold(code: str, reason: str, proposal: Any | None = None, *, regime: str | None = None) -> CPRAgentOutcome: @@ -442,14 +480,21 @@ def decide( started = monotonic() executor = ThreadPoolExecutor(max_workers=1) release_when_finished = False + # The parent may reach its wall-clock deadline while the child thread is + # still unwinding. This shared, append-only audit lets that fail-closed + # path retain every completed attempt rather than returning a blank HOLD. + attempt_evidence: list[CPRAttemptEvidence] = [] try: future = executor.submit( self._run_turn_with_tool_retry, context, bar_signature, + started + self.timeout_seconds, + current_signature, + attempt_evidence, ) try: - result, inference_attempts, combined_usage = future.result( + result, recorded_attempts, combined_usage, terminal_code = future.result( timeout=self.timeout_seconds ) except TimeoutError: @@ -459,35 +504,57 @@ def decide( future.cancel() release_when_finished = True future.add_done_callback(lambda _future: self._inference_lock.release()) - return _hold("timeout", "Codex did not finish inside the configured deadline.") + return self._terminal_hold_with_attempt_evidence( + "timeout", attempt_evidence, current_signature + ) except Exception as error: # optional SDK failure must disable this agent only - return _hold("runtime_error", f"Optional Codex runtime failed: {type(error).__name__}.") + # Timeout-class errors retain their conservative pre-launch audit. + # Other first-turn runtime errors keep the established generic + # runtime outcome and never expose an exception message. + if self._is_timeout_error(error): + return self._terminal_hold_with_attempt_evidence( + "timeout", attempt_evidence, current_signature + ) + return self._terminal_hold_with_attempt_evidence( + "runtime_error", attempt_evidence, current_signature + ) finally: executor.shutdown(wait=False, cancel_futures=True) if not release_when_finished: self._inference_lock.release() latency_ms = int((monotonic() - started) * 1000) - outcome = self._validate_run(result, context, bar_signature, current_signature) + outcome = ( + self._terminal_hold_with_attempt_evidence( + terminal_code, list(recorded_attempts), current_signature + ) + if terminal_code is not None + else self._validate_run(result, context, bar_signature, current_signature) + ) outcome.latency_ms = latency_ms outcome.token_usage = combined_usage outcome.tool_evidence = result.tool_calls - outcome.inference_attempts = inference_attempts + outcome.inference_attempts = len(recorded_attempts) + outcome.attempt_evidence = recorded_attempts return outcome def _run_turn_with_tool_retry( self, context: Mapping[str, Any], bar_signature: str, - ) -> tuple[CPRAgentRunResult, int, dict[str, int]]: + deadline: float, + current_signature: Callable[[], str] | None, + attempt_evidence: list[CPRAttemptEvidence], + ) -> tuple[CPRAgentRunResult, tuple[CPRAttemptEvidence, ...], dict[str, int], str | None]: """Retry incomplete frozen-tool evidence once inside one total deadline. Both attempts receive the same in-memory context and bar signature. A retry therefore cannot silently move to newer market facts. The second - attempt receives only the wall-clock budget left after the first one, - so this recovery path never doubles the configured SDK timeout. + attempt receives only the wall-clock budget left after the first one. + That shared budget includes parent/child process setup as well as the + SDK turn itself, so this recovery path never doubles the configured + CPR inference timeout. """ - deadline = monotonic() + self.timeout_seconds results: list[CPRAgentRunResult] = [] for attempt_index in range(2): if attempt_index == 0: @@ -498,17 +565,90 @@ def _run_turn_with_tool_retry( else: attempt_timeout = deadline - monotonic() if attempt_timeout <= 0: - break - result = self._run_turn( - context, - bar_signature, - timeout_seconds=attempt_timeout, + # The repair was selected but no wall-clock budget remains. + # Record that fact explicitly without inventing a tool call + # or token count for a child process that never launched. + attempt_evidence.append( + CPRAttemptEvidence( + attempt_number=2, + request_kind=CPRTurnRequestKind.TOOL_REPAIR, + evidence_code="timeout", + tool_records=(), + token_usage={}, + ) + ) + return results[-1], tuple(attempt_evidence), self._combined_token_usage(results), "timeout" + request_kind = ( + CPRTurnRequestKind.NORMAL + if attempt_index == 0 + else CPRTurnRequestKind.TOOL_REPAIR ) + # Install a conservative no-evidence marker before the optional + # runtime starts. If the parent deadline wins the race, this still + # records which turn slot was selected. Empty tool/usage fields mean + # the child returned no provable evidence; they are not an assertion + # that a launched child consumed exactly zero tokens. + diagnostic_index = len(attempt_evidence) + attempt_evidence.append( + CPRAttemptEvidence( + attempt_number=attempt_index + 1, + request_kind=request_kind, + evidence_code="timeout", + tool_records=(), + token_usage={}, + ) + ) + try: + result = self._run_turn( + context, + bar_signature, + request_kind=request_kind, + timeout_seconds=attempt_timeout, + ) + except Exception as error: + # Preserve the usual first-turn exception behavior. Once the + # normal result proved a repair was needed, however, a failed + # repair must retain that completed first audit trail. + if attempt_index == 0: + if not self._is_timeout_error(error): + # A normal child failed before returning any evidence. + # Keep one typed record without inventing tools, usage, + # or an exception string for the operator audit. + attempt_evidence[diagnostic_index] = CPRAttemptEvidence( + attempt_number=1, + request_kind=CPRTurnRequestKind.NORMAL, + evidence_code="runtime_error", + tool_records=(), + token_usage={}, + ) + raise + terminal_code = "timeout" if self._is_timeout_error(error) else "runtime_error" + attempt_evidence[diagnostic_index] = ( + CPRAttemptEvidence( + attempt_number=2, + request_kind=CPRTurnRequestKind.TOOL_REPAIR, + evidence_code=terminal_code, + tool_records=(), + token_usage={}, + ) + ) + return results[-1], tuple(attempt_evidence), self._combined_token_usage(results), terminal_code results.append(result) evidence_error = self._tool_evidence_error(result) - if ( - evidence_error is None - or evidence_error[0] not in _RETRIABLE_TOOL_EVIDENCE_CODES + completed_evidence = CPRAttemptEvidence( + attempt_number=attempt_index + 1, + request_kind=request_kind, + evidence_code=None if evidence_error is None else evidence_error[0], + tool_records=result.tool_calls, + token_usage=dict(result.token_usage), + ) + attempt_evidence[diagnostic_index] = completed_evidence + if not self._retry_is_allowed( + result, + context, + bar_signature, + current_signature, + evidence_error, ): break @@ -516,7 +656,114 @@ def _run_turn_with_tool_retry( # this list cannot be empty. Keeping the assertion documents that local # invariant without converting an SDK exception into trusted evidence. assert results - return results[-1], len(results), self._combined_token_usage(results) + return results[-1], tuple(attempt_evidence), self._combined_token_usage(results), None + + @staticmethod + def _is_timeout_error(error: Exception) -> bool: + """Recognize local deadline exceptions without retaining their details. + + The optional adapter may surface either the standard library's + ``TimeoutExpired`` or the executor's ``TimeoutError``. The host stores + only the safe classification, never a potentially sensitive message. + """ + + return isinstance(error, TimeoutError) or type(error).__name__ == "TimeoutExpired" + + @staticmethod + def _terminal_failure_reason(code: str) -> str: + """Return a credential-safe reason for a terminal Codex-turn outcome. + + Both the initial turn and an optional repair use this helper. The + historical human-readable text uses the word ``corrective`` for either + path, so ``attempt_evidence.request_kind`` is the authoritative field + when an operator needs to distinguish ``normal`` from ``tool_repair``. + The reason itself never includes exception text, child output, command + arguments, or local paths. + """ + + if code == "timeout": + return "The corrective Codex turn exhausted the original deadline." + return "The optional corrective Codex runtime failed." + + def _terminal_hold_with_attempt_evidence( + self, + code: str, + attempt_evidence: list[CPRAttemptEvidence], + current_signature: Callable[[], str] | None, + ) -> CPRAgentOutcome: + """Build a terminal HOLD with its one retained host signature sample. + + These paths never reach ``_validate_run`` because the initial or repair + child timed out or failed. Capture one signature here rather than + leaving a JSONL row ambiguous or calling the mutable shared-data getter + later. The call happens once during finalization, so the audit describes + this decision even if the market store advances immediately afterward. + """ + + recorded_attempts = tuple(attempt_evidence) + outcome = _hold(code, self._terminal_failure_reason(code)) + outcome.inference_attempts = len(recorded_attempts) + outcome.attempt_evidence = recorded_attempts + outcome.token_usage = self._combined_attempt_token_usage(recorded_attempts) + outcome.tool_evidence = next( + (record.tool_records for record in reversed(recorded_attempts) if record.tool_records), + (), + ) + return self._with_validation_current_signature( + outcome, + current_signature() if current_signature is not None else None, + ) + + @staticmethod + def _combined_attempt_token_usage( + attempts: tuple[CPRAttemptEvidence, ...], + ) -> dict[str, int]: + """Combine retained attempt counters with the same context-window rule.""" + + combined: dict[str, int] = {} + for attempt in attempts: + for key, value in attempt.token_usage.items(): + if key == "model_context_window": + combined[key] = max(combined.get(key, 0), int(value)) + else: + combined[key] = combined.get(key, 0) + int(value) + return combined + + def _retry_is_allowed( + self, + result: CPRAgentRunResult, + context: Mapping[str, Any], + bar_signature: str, + current_signature: Callable[[], str] | None, + evidence_error: tuple[str, str] | None, + ) -> bool: + """Allow the one repair only for otherwise-valid missing/failed reads. + + A repair exists to recover an incomplete observation of immutable MCP + facts. It must not hide a stale bar, bad schema/model/prompt echo, or + deterministic host-policy rejection behind a second model attempt. + """ + + if evidence_error is None or evidence_error[0] not in _RETRIABLE_TOOL_EVIDENCE_CODES: + return False + if current_signature is not None and current_signature() != bar_signature: + return False + try: + from cpr_ai_schema import CPRAgentDecision + + proposal = CPRAgentDecision.model_validate_json(result.final_response) + except Exception: + return False + if proposal.model_used != self.model: + return False + expected_prompt = self.prompt_version + if expected_prompt is None: + from cpr_ai_prompt import CPR_AI_PROMPT_VERSION + + expected_prompt = CPR_AI_PROMPT_VERSION + if proposal.prompt_version != expected_prompt: + return False + return self.policy.validate(context, proposal).accepted @staticmethod def _combined_token_usage( @@ -543,6 +790,7 @@ def _run_turn( context: Mapping[str, Any], bar_signature: str, *, + request_kind: CPRTurnRequestKind = CPRTurnRequestKind.NORMAL, timeout_seconds: float | None = None, ) -> CPRAgentRunResult: """Build prompt/schema lazily and pass only advisory inputs to the child. @@ -566,6 +814,7 @@ def _run_turn( reasoning_effort=self.reasoning_effort, prompt_version=self.prompt_version or CPR_AI_PROMPT_VERSION, output_schema=CPRAgentDecision.model_json_schema(), + request_kind=request_kind, # Direct diagnostic callers historically used this helper without # an explicit deadline. Normal and retry paths pass their exact # per-attempt budget; the fallback preserves that diagnostic API. @@ -592,26 +841,49 @@ def _validate_run( to place a trade. """ + # Sample mutable market identity exactly once for this validation. The + # audit row must describe this host decision, not a later poll that may + # have received an official-candle correction in the meantime. + validation_current_signature = ( + current_signature() if current_signature is not None else None + ) evidence_error = self._tool_evidence_error(result) if evidence_error is not None: - return _hold(*evidence_error) - if current_signature is not None and current_signature() != bar_signature: - return _hold("stale_bar_signature", "The frozen completed bar is no longer current.") + return self._with_validation_current_signature( + _hold(*evidence_error), validation_current_signature + ) + if ( + current_signature is not None + and validation_current_signature != bar_signature + ): + return self._with_validation_current_signature( + _hold("stale_bar_signature", "The frozen completed bar is no longer current."), + validation_current_signature, + ) try: from cpr_ai_schema import CPRAgentDecision proposal = CPRAgentDecision.model_validate_json(result.final_response) except Exception: - return _hold("malformed_output", "Codex output did not match the strict decision schema.") + return self._with_validation_current_signature( + _hold("malformed_output", "Codex output did not match the strict decision schema."), + validation_current_signature, + ) if proposal.model_used != self.model: - return _hold("model_mismatch", "Model echo does not match the configured model.", proposal) + return self._with_validation_current_signature( + _hold("model_mismatch", "Model echo does not match the configured model.", proposal), + validation_current_signature, + ) expected_prompt = self.prompt_version if expected_prompt is None: from cpr_ai_prompt import CPR_AI_PROMPT_VERSION expected_prompt = CPR_AI_PROMPT_VERSION if proposal.prompt_version != expected_prompt: - return _hold("prompt_version_mismatch", "Prompt-version echo does not match the host prompt.", proposal) + return self._with_validation_current_signature( + _hold("prompt_version_mismatch", "Prompt-version echo does not match the host prompt.", proposal), + validation_current_signature, + ) outcome = self.policy.validate(context, proposal) # The SDK boundary has proved that this was a contemporaneous, pinned # regime classification. Preserve it even when hard execution gates @@ -621,6 +893,18 @@ def _validate_run( "invalid_frozen_context", }: outcome.accepted_regime = proposal.regime + return self._with_validation_current_signature( + outcome, validation_current_signature + ) + + @staticmethod + def _with_validation_current_signature( + outcome: CPRAgentOutcome, + validation_current_signature: str | None, + ) -> CPRAgentOutcome: + """Retain the one validation-time signature on every audited outcome.""" + + outcome.validation_current_signature = validation_current_signature return outcome @staticmethod @@ -647,4 +931,12 @@ def _tool_evidence_error(result: CPRAgentRunResult) -> tuple[str, str] | None: return None -__all__ = ["CPRAgent", "CPRAgentOutcome", "CPRAgentRunResult", "CPRHostPolicy", "CPRToolCallRecord"] +__all__ = [ + "CPRAgent", + "CPRAgentOutcome", + "CPRAgentRunResult", + "CPRAttemptEvidence", + "CPRHostPolicy", + "CPRToolCallRecord", + "CPRTurnRequestKind", +] diff --git a/Signal Generators/CPR AI Agent/cpr_ai_codex_runner.py b/Signal Generators/CPR AI Agent/cpr_ai_codex_runner.py index 38f7fa6..00d9609 100644 --- a/Signal Generators/CPR AI Agent/cpr_ai_codex_runner.py +++ b/Signal Generators/CPR AI Agent/cpr_ai_codex_runner.py @@ -29,7 +29,7 @@ from pathlib import Path from typing import Any -from cpr_ai_agent import CPRAgentRunResult, CPRToolCallRecord +from cpr_ai_agent import CPRAgentRunResult, CPRToolCallRecord, CPRTurnRequestKind from cpr_ai_codex_subprocess import build_isolated_thread_config _PROCESS_CODEX_HOME_LOCK = threading.Lock() @@ -171,6 +171,9 @@ def run_codex_turn(**kwargs: Any) -> CPRAgentRunResult: context = kwargs.get("context") if not isinstance(context, Mapping): raise ValueError("Codex turn requires a frozen CPR context mapping.") + request_kind = kwargs.get("request_kind", CPRTurnRequestKind.NORMAL) + if not isinstance(request_kind, CPRTurnRequestKind): + raise ValueError("Codex turn request kind must be a CPRTurnRequestKind enum value.") timeout_seconds = float(kwargs.get("timeout_seconds", 90.0)) if not math.isfinite(timeout_seconds) or timeout_seconds <= 0.0: raise ValueError("Codex subprocess timeout must be a positive finite number.") @@ -198,6 +201,10 @@ def run_codex_turn(**kwargs: Any) -> CPRAgentRunResult: "reasoning_effort": kwargs.get("reasoning_effort"), "prompt": kwargs.get("prompt"), "output_schema": kwargs.get("output_schema"), + # This enum is the only parent-to-child authority for selecting a + # turn request. The child maps it to one constant; arbitrary text + # can never become a repair prompt. + "request_kind": request_kind.value, } completed = subprocess.run( [sys.executable, str(script)], diff --git a/Signal Generators/CPR AI Agent/cpr_ai_codex_subprocess.py b/Signal Generators/CPR AI Agent/cpr_ai_codex_subprocess.py index ae4cd1d..ae11ce4 100644 --- a/Signal Generators/CPR AI Agent/cpr_ai_codex_subprocess.py +++ b/Signal Generators/CPR AI Agent/cpr_ai_codex_subprocess.py @@ -19,7 +19,7 @@ from pathlib import Path from typing import Any -_REQUEST_KEYS = {"snapshot_path", "model", "reasoning_effort", "prompt", "output_schema"} +_REQUEST_KEYS = {"snapshot_path", "model", "reasoning_effort", "prompt", "output_schema", "request_kind"} _EXPECTED_TOOLS = ("session_levels", "momentum_vwap", "market_structure", "position_state") # ``build_system_prompt()`` remains the durable policy authority. Repeating the # four reads in the immediate turn request is intentional defense in depth: a @@ -30,6 +30,16 @@ "momentum_vwap, market_structure, and position_state. Wait for all four " "calls to complete, then evaluate the frozen CPR context and return one decision." ) +# This deliberately constant repair request is selected solely by the parent +# enum. It supplies feedback about evidence rejection without allowing either +# a caller or the first model response to inject a new instruction. +_TOOL_REPAIR_TURN_REQUEST = ( + "The prior attempt was rejected because required frozen-tool evidence was missing or failed. " + "Before returning JSON, call each frozen MCP tool exactly once: session_levels, momentum_vwap, " + "market_structure, and position_state. Wait for all four calls to complete, then evaluate the same frozen " + "CPR context and return one decision." +) +_TURN_REQUESTS = {"normal": _TURN_REQUEST, "tool_repair": _TOOL_REPAIR_TURN_REQUEST} _ALLOWED_TURN_ITEM_TYPES = frozenset( { # The SDK records the prompt submitted through ``Thread.run`` as a @@ -166,6 +176,9 @@ def _run_request(request: Mapping[str, Any]) -> dict[str, Any]: from openai_codex import ApprovalMode, Codex, Sandbox + request_kind = request.get("request_kind") + if not isinstance(request_kind, str) or request_kind not in _TURN_REQUESTS: + raise ValueError("Invalid isolated Codex request kind.") snapshot_path = str(request["snapshot_path"]) runtime_directory = str(Path(snapshot_path).parent) config = build_isolated_thread_config(snapshot_path) @@ -183,7 +196,7 @@ def _run_request(request: Mapping[str, Any]) -> dict[str, Any]: approval_mode=ApprovalMode.deny_all, ) result = thread.run( - _TURN_REQUEST, + _TURN_REQUESTS[request_kind], approval_mode=ApprovalMode.deny_all, output_schema=request["output_schema"], effort=request["reasoning_effort"], @@ -213,7 +226,11 @@ def main() -> int: try: request = json.load(sys.stdin) - if not isinstance(request, Mapping) or set(request) != _REQUEST_KEYS: + if ( + not isinstance(request, Mapping) + or set(request) != _REQUEST_KEYS + or request.get("request_kind") not in _TURN_REQUESTS + ): raise ValueError("Invalid isolated Codex request.") response = _run_request(request) except (ImportError, KeyError, TypeError, ValueError, RuntimeError) as error: diff --git a/Signal Generators/CPR AI Agent/cpr_ai_decision_log.py b/Signal Generators/CPR AI Agent/cpr_ai_decision_log.py index 1b46dc1..4c3a090 100644 --- a/Signal Generators/CPR AI Agent/cpr_ai_decision_log.py +++ b/Signal Generators/CPR AI Agent/cpr_ai_decision_log.py @@ -16,8 +16,16 @@ import json import re from collections.abc import Mapping +from datetime import datetime from pathlib import Path from typing import Any +from zoneinfo import ZoneInfo + +from cpr_ai_tools import EXPECTED_TOOL_NAMES + +# The isolated adapter reports terminal evidence only for these statuses. Keep +# the serializer closed so untrusted child text cannot become a JSONL disclosure. +_SAFE_TOOL_STATUSES = frozenset({"completed", "failed"}) _SENSITIVE_KEY_TOKENS = frozenset( { @@ -30,13 +38,21 @@ "brokers", "credential", "credentials", + "path", + "paths", "order", "orders", "password", "passwords", "secret", "secrets", + "symbol", + "symbols", "token", + "quantity", + "quantities", + "reasoning", + "response", "venue", "venues", } @@ -130,18 +146,29 @@ def write( token_usage: Mapping[str, Any], tool_evidence: list[Mapping[str, Any]], execution: Mapping[str, Any] | None = None, + audit_stage: str = "DIRECT", + bar_metadata: Mapping[str, Any] | None = None, ) -> None: """Append one complete sanitized record after a host decision. - The default execution object explicitly says no order was submitted, - which is safer than leaving an absent field open to interpretation. + ``DIRECT`` is a safe default for older diagnostics that call this + logger outside the worker. The CPR worker labels its first record + ``PRE_ACTION`` and its follow-up provenance record ``POST_ACTION``. Parent directories are created only after the enabled guard passes. """ if not self.enabled: return + bar = dict(bar_metadata or {}) + # The outcome is the authority for this value: it was sampled once at + # validation and must not be rebuilt from mutable shared market data. + bar["validation_current_signature"] = getattr( + outcome, "validation_current_signature", None + ) row = _sanitized( { + "recorded_at": datetime.now(ZoneInfo("Asia/Kolkata")).isoformat(), + "audit_stage": audit_stage, "frozen_context": frozen_context, "proposal": proposal, "accepted_regime": outcome.accepted_regime, @@ -167,12 +194,69 @@ def write( # totals understandable during later operational review. "inference_attempts": int(getattr(outcome, "inference_attempts", 1)), "token_usage": token_usage, - "tool_evidence": tool_evidence, + "tool_evidence": self._safe_tool_records(tool_evidence), + "attempt_evidence": self._safe_attempt_evidence( + getattr(outcome, "attempt_evidence", ()) + ), + "bar": bar, } ) self.path.parent.mkdir(parents=True, exist_ok=True) with self.path.open("a", encoding="utf-8") as handle: handle.write(json.dumps(row, sort_keys=True, separators=(",", ":")) + "\n") + @staticmethod + def _safe_tool_records(records: Any) -> list[dict[str, str]]: + """Keep only canonical MCP tool names and terminal statuses. + + SDK error text can contain a local path, a response body, or credentials. + The typed evidence code already says whether a tool failed, so retaining + opaque error text is unnecessary and unsafe for an operational audit. + """ + + safe_records: list[dict[str, str]] = [] + for record in records: + if isinstance(record, Mapping): + tool = record.get("tool") + status = record.get("status") + else: + tool = getattr(record, "tool", None) + status = getattr(record, "status", None) + if tool not in EXPECTED_TOOL_NAMES or status not in _SAFE_TOOL_STATUSES: + # The typed evidence code still records why this attempt was + # rejected, so retain no arbitrary tool/status payload at all. + continue + safe_records.append({"tool": tool, "status": status}) + return safe_records + + @classmethod + def _safe_attempt_evidence(cls, attempts: Any) -> list[dict[str, Any]]: + """Serialize Task 2 attempt facts without model text or SDK errors.""" + + safe_attempts: list[dict[str, Any]] = [] + for attempt in attempts: + if isinstance(attempt, Mapping): + attempt_number = attempt.get("attempt_number", 0) + request_kind = attempt.get("request_kind") + evidence_code = attempt.get("evidence_code") + tool_records = attempt.get("tool_records", ()) + token_usage = attempt.get("token_usage", {}) + else: + attempt_number = getattr(attempt, "attempt_number", 0) + request_kind = getattr(attempt, "request_kind", None) + evidence_code = getattr(attempt, "evidence_code", None) + tool_records = getattr(attempt, "tool_records", ()) + token_usage = getattr(attempt, "token_usage", {}) + safe_attempts.append( + { + "attempt_number": int(attempt_number), + "request_kind": str(request_kind), + "evidence_code": evidence_code, + "tool_records": cls._safe_tool_records(tool_records), + "token_usage": dict(token_usage), + } + ) + return safe_attempts + __all__ = ["CPRDecisionLogger"] diff --git a/Signal Generators/SL Hunting AI Agent/premarket_note.json b/Signal Generators/SL Hunting AI Agent/premarket_note.json index 2fa6065..cf5c272 100644 --- a/Signal Generators/SL Hunting AI Agent/premarket_note.json +++ b/Signal Generators/SL Hunting AI Agent/premarket_note.json @@ -1,26 +1,25 @@ { - "for_date": "2026-08-13", - "source": "Intraday Hunter, 'Prediction For 13 AUG 2026' (PfthlsdW2E8, uploaded 2026-08-12)", - "context": "Sensex sold hard, then took support EXACTLY at the round number and rallied, chasing out everyone who was short. BankNIFTY has been positive from the start. NIFTY sold but is recovering. BUYERS, not sellers, are now the seated crowd.", + "for_date": "2026-08-14", + "source": "Intraday Hunter, 'Prediction For 14 AUG 2026' (MhmlrlUEUGI, uploaded 2026-08-13)", + "context": "No decisive crowd anywhere. BankNIFTY made little momentum with buyers and sellers at the SAME price level; Sensex ran both ways. The market held its round number and recovered, so sellers are unlikely to have carried overnight.", "plan": [ - "FLAT to GAP-DOWN: identify SELL-side setups. Stated for all three indices, and on BankNIFTY the reason is explicit -- target the BUYERS who are now seated.", - "GAP-UP above the round number: go WITH the market and identify BUY-side setups. Do NOT try to hunt the buyers there.", - "The reason is WHERE the stops sit: 'buyers' SLs should be BELOW the round number.' A gap-up above it leaves their stops far away, so they are not huntable and following beats fading.", - "The seller crowd is SPENT. Sensex took support at the round number and the recovery chased shorts out on the retracement -- the same spent-seller condition as 12 Aug.", - "BankNIFTY is the index where buyers are genuinely seated: positive from the open, a retracement, then more upside. That is the crowd a gap-down would hunt.", - "SENSEX EXPIRY tomorrow, flagged explicitly. Expect expiry pinning and premium distortion on the Sensex read specifically.", - "NIFTY is the least committed of the three: selling, but 'some recovery is also visible'. Treat its conditional as weaker than BankNIFTY's." + "FLAT to GAP-DOWN: identify SELL-side setups. On Sensex the reason is explicit -- target the BUYERS, since the market has sat on support since yesterday and the sellers did not hold.", + "GAP-UP: go WITH the market and identify BUY-side setups. On BankNIFTY 'the buyer becomes safe' there, so there is nothing left to hunt on that side.", + "NOBODY IS DECISIVELY SEATED -- on BankNIFTY 'buyers and sellers would be sitting at the SAME price level', and on Sensex claiming either side is seated 'would be WRONG'. This is the empty-book condition.", + "His mechanism for that condition, worth keeping: 'where the crowd is THIN, that is where the market tries to make momentum.' Thin is not dead -- the move goes where there is least resistance.", + "He PREFERS a gap to a flat open: 'better it is not flat -- either a gap-up or a gap-down.' A flat open 'extracts a small momentum and goes away', producing nothing worth trading.", + "There is little pressure on BankNIFTY's sellers, so the market cannot target them directly. The flat/gap-down branch is therefore a FOLLOW of the drift, not a seller hunt." ], "levels": [ { "index": "NIFTY", - "resistance": [24500, 24600], - "support": [24260] + "resistance": [24440, 24540], + "support": [24275, 24176] }, { "index": "BANKNIFTY", - "resistance": [58000, 58300], - "support": [57500, 57310] + "resistance": [57890, 58000], + "support": [57500, 57320] }, { "index": "SENSEX", diff --git a/Signal Generators/SL Hunting AI Agent/sl_hunting_doc.md b/Signal Generators/SL Hunting AI Agent/sl_hunting_doc.md index 5fdb21d..063e694 100644 --- a/Signal Generators/SL Hunting AI Agent/sl_hunting_doc.md +++ b/Signal Generators/SL Hunting AI Agent/sl_hunting_doc.md @@ -3258,7 +3258,210 @@ a missing advisory level is safer than a wrong one, and the test asserts NIFTY carries exactly one support so a later "tidy-up" cannot invent the second. Sensex's 78145 resistance is recorded as heard. +> **Corrected 2026-08-13 (see the 14 Aug entry):** "2476" was **24,176**, read +> directly off the 1080p chart frame. Omitting was the right call given only the +> caption, but reading the frame is better still, and is now the standard step +> whenever a level does not parse — the chart is authoritative, the auto-caption +> is not. + Test updated: `test_shipped_note_matches_august_13_intraday_hunter_plan` replaces the 12 Aug equivalent and asserts both branch directions plus the round-number stop-location reasoning, because an inverted plan and a smoothed- away justification are the two failures a copy-forward would produce. + +--- + +## Video addendum - the 13 Aug LIVE SESSION (v4g) + +**Source:** Intraday Hunter live session, 13 Aug 2026 (`Vig9Kjab2T0`, 10:36, +published 11:36 IST). A win, but a REDUCED one - the trade reached most of its +target, reversed into a loss, recovered, and was booked for roughly half. That +arc is what makes the session valuable: it produces the two structural ideas +below and the most complete discipline block in the series. + +### A completed stop-hunt ends that direction + +The sharpest new idea, and it inverts the naive reading of a bounce: + +> "Yesterday the market gave good selling, then took support EXACTLY at the 500 +> level and gave a retracement. Because of that retracement, whoever was selling +> got chased out... so the chances of going DIRECTLY UP are LOW." + +The up-move existed *to clear the shorts*. With the shorts gone there is nobody +left to squeeze, so the bounce has spent its fuel. His operating rule: + +> "If it has chased the sellers out, we try to follow THAT SAME DIRECTION." + +- meaning the direction that preceded the clearing bounce, not the bounce. The +escape hatch is explicit and is kept in the prose: a fresh large gap recruits a +new crowd and restarts the question. + +### The round number is where the thesis DIES + +v4d used round numbers to place targets; v4c used them to explain recruitment. +This adds the third and most operational use - the declared invalidation: + +> "Until the market crosses the round number - as we see in NIFTY, the 24,500 +> level - we will not have much problem." +> "When is there no danger to these buyers? If the market goes above 58,000." + +Each index carries its own named level, decided before entry, and a decisive +cross means the read has failed even if the arithmetic stop is untouched. + +### The discipline block, and why it needed a test + +> "Before the target is hit there is a fear - should I book here, what if the +> market turns? **Fear is not a big deal, I feel it too.** But do NOT convert +> that fear into ACTION." +> "If you cut early it gradually becomes a HABIT. Then you cut small profits and +> leave, and when there is a loss you wait a long time to save the position and +> take a BIG loss. That is why most traders never become profitable." + +This sits directly on top of v4f's BOOK WHEN THE PROFIT STOPS GROWING, and the +pair is the most dangerous in the prompt: read carelessly, v4g reads as "hold +through everything" and disables v4f, while v4f licenses exactly the fear-driven +exit v4g forbids. The distinction encoded in both directions is **measurement +versus emotion** - the profit RATE falling is a measurement; "it might turn" is +not. `test_v4g_fear_rule_and_v4f_book_rule_do_not_cancel_each_other` asserts +both halves say so, and that the fear rule still enumerates the legitimate exit +reasons rather than banning exits. + +Two more from the same arc: **time spent shrinks the achievable target** ("the +market turned and wasted our TIME... only half the profit is showing, where +earlier it showed double"), and **never exit at zero after a good profit has +printed** - once real open profit has appeared, the floor stops being breakeven. + +### Knowledge changes (v4g, all prose) + +- `OPENING_DRIVE`: A COMPLETED STOP-HUNT ENDS THAT DIRECTION; THE ROUND NUMBER + IS WHERE THE THESIS DIES, NOT JUST WHERE IT PAYS. +- `RISK`: FEAR IS NOT A SIGNAL - NEVER CONVERT IT INTO AN EXIT; TIME SPENT IN + THE TRADE SHRINKS THE ACHIEVABLE TARGET; NEVER EXIT AT ZERO AFTER A GOOD + PROFIT HAS PRINTED. +- Test markers: `test_system_prompt_has_v4g_stop_hunt_completion_and_discipline_knowledge` + plus two drift guards - the fear/book pair above, and + `test_v4g_stop_hunt_rule_does_not_become_always_fade_the_bounce`, which keeps + the large-gap escape hatch alive so the rule cannot harden into "always fade". +- Prompt size 105,933 -> 111,283 chars. **Headroom is now 8,717** - at the + recent ~5,000 chars per version that is roughly two more before the 120,000 + cap, so the next pass should start pruning superseded prose rather than only + appending. + +### How our agent traded the same session + +**Provisional - the runner was still live at 14:13.** Realized so far: +**-9,229.75** across 45 legs, and the shape is the exact inverse of 12 Aug. + +| Strategy | Legs | Realized | +|---|---|---| +| SL Hunting AI | 4 | **+2,727.00** | +| Heikin Ashi | 6 | +2,431.00 | +| Supertrend Bullish | 1 | +1,043.25 | +| RSI Reversal | 1 | +994.50 | +| SMA Crossover | 2 | +539.50 | +| Long Strangle | 8 | +305.50 | +| ... | | | +| CPR Algo 3 | 1 | -2,158.00 | +| Donchian Bearish | 1 | -2,567.50 | +| Supertrend | 4 | -2,671.50 | +| Opening Strike | 1 | -2,717.00 | +| Renko | 6 | -4,127.50 | +| **Total** | **45** | **-9,229.75** | + +**SL Hunting was the best strategy on the board**, cross-checked against its own +`Result summary` (+2,727.00, Trades=2). Yesterday it was the worst. + +What changed is exactly what v4g and v4d describe. Both trades were SHORT off the +flat open - the same side IH took - and **both exits keyed off the round number**: + +| Entry | Setup | Exit | Held | +|---|---|---|---| +| 09:29 | flat_open_pivot_breakdown_bearish_engulfing (stop 24353, target 24300) | 09:35 `profit_booking_round_number` | 6 min | +| 10:06 | double_top_rejection_confirmed_bearish | 10:26 `profit_book_stall_near_round_number` | 20 min | + +Contrast with 12 Aug: four entries in 47 minutes, three released on premise-STALL +judgements, one cut by the hierarchy rule after 60 seconds. Today: two entries in +57 minutes, both booked on a NAMED level. The difference is not the read - it was +short both days - it is that the exits had a checkable reason, which is precisely +what v4g's FEAR IS NOT A SIGNAL demands and what the stall-churn lacked. + +One caveat against reading too much into it: two trades is a small sample, and +the deterministic strategies had a poor day on the same tape (Renko -4,127.50 over +six trades), so the basket is deeply negative regardless. + +--- + +### Pre-open note for 2026-08-14 (Friday) + +**Source:** Intraday Hunter, "Prediction For 14 AUG 2026" (`MhmlrlUEUGI`, +uploaded 2026-08-13, 2:14). Note-only; no knowledge version attached. + +**Neither side is seated, and he says so for two of the three indices.** This is +the empty-book condition v4f describes, stated more plainly than the series has +had it before: + +> BankNIFTY: "we did not see much momentum... buyers and sellers would be sitting +> at the SAME price level. Not many people held their positions." +> Sensex: "momentum on both sides remained -- a bit of rejection, a bit of +> buying. So to say more sellers are seated, or more buyers are seated, would be +> WRONG." + +That matters because v4e's expensive lesson was forecasting a crowd into +existence on exactly this condition, and v4f's answer was to wait for +confirmation. The note records the absence rather than smoothing it into a +direction. + +**His mechanism for trading a thin book** is the line worth keeping: + +> "If they did not hold positions, then WHERE THE CROWD IS THIN, THAT IS WHERE +> THE MARKET TRIES TO MAKE MOMENTUM." + +Thin is not dead: the move goes where there is least resistance, which is a +different claim from "no crowd means no trade" and sits alongside v4f's +AN EMPTY BOOK MEANS A TRAP IS COMING rather than against it. + +**He prefers a gap to a flat open, explicitly.** New for the series as a stated +preference rather than an inference: + +> "Better that it is NOT flat -- either a gap-up or a gap-down would be better, +> because in flat it keeps extracting a small momentum and going away. It does +> not make any special momentum." + +That is v4d's participation reading turned into an operational preference: a +flat open grants everyone entry, so it produces chop rather than a move. + +Plan: SELL-side on flat-to-gap-down (on Sensex explicitly to target the buyers +sitting on support since yesterday), BUY-side and follow on a gap-up, where "the +buyer becomes safe" and there is nothing left to hunt. On BankNIFTY he notes +there is little pressure on the sellers, so that branch is a **follow of the +drift rather than a seller hunt** -- a distinction the note keeps. + +**The garbled level was RECOVERED FROM THE VIDEO, and this replaces omitting.** +NIFTY's second support arrived as the same "2476" token as the 13 Aug video — +twice in two days, so a consistent ASR failure on one number rather than noise. +Instead of dropping it again, the frame at 2:04 was read directly: force the +player to 1080p (`setPlaybackQualityRange`), seek, draw the `