references/gate-decision-table.md
# Contrarian Setup Gate -- Decision Table
## Purpose
This gate combines three normalized states -- crowding (C), news failure (N), and price action (P) -- into one `setup_status` via SEQUENTIAL pipeline-order evaluation. Each of N and P is one of five states: `CONFIRMED`, `NOT_CONFIRMED`, `INSUFFICIENT`, `PENDING` (the report file was not provided), or `INVALID` (a report was provided but is unusable). C shares the same five states, except it is never `PENDING` -- the detector report is always required.
Each step is evaluated **in pipeline order** -- crowding, then news, then price-action -- and each step FULLY SETTLES (reaches a determination) before the next step's state is even inspected. This document is the authoritative spec the test suite (`scripts/tests/test_gate_logic.py`) checks against.
## Why sequential, not "any-X across both downstream steps" (v4, PR #249 P1-3)
An earlier design (v3) evaluated the two downstream steps AS A GROUP: "any provided N or P that is INVALID" was checked before "any N or P that is NOT_CONFIRMED," scanning both steps for each condition before moving to the next condition. That aggregation let a LATER step's problem soften an EARLIER step's definitive verdict -- e.g. news already NOT_CONFIRMED (a hard rejection) got downgraded to INSUFFICIENT_EVIDENCE merely because price-action's file happened to be corrupted, even though price-action was never going to be consulted once news had already rejected the setup. This is the same softening-principle violation already fixed twice elsewhere in this gate (the loader path in v1, the consistency path in v2/v3) -- now caught a third time, in cross-step aggregation, by an independent user review. The sequential model makes the principle structural: once a step settles, later steps simply aren't looked at for that decision.
## Precedence Rules
1. **Crowding, evaluated first and exclusively.**
- C is `INVALID` or `INSUFFICIENT` -> `INSUFFICIENT_EVIDENCE`.
- C is `NOT_CONFIRMED` (classification `NEUTRAL`) -> `REJECTED`, **regardless of N or P's state** -- N/P are never even inspected for this decision. N/P are still echoed in the `inputs` audit block, but never change this status.
- C is `CONFIRMED` -> continue to step 2.
2. **News, evaluated next -- if provided (state != `PENDING`).** The same four-way settlement: `INVALID` -> `INSUFFICIENT_EVIDENCE`; `NOT_CONFIRMED` -> `REJECTED`; `INSUFFICIENT` -> `INSUFFICIENT_EVIDENCE`; `CONFIRMED` -> continue to step 3. **Price-action is not inspected at all until news has settled as `CONFIRMED`** (or is `PENDING`, handled by step 3's out-of-order case below).
3. **Price-action, evaluated last.** If news was `CONFIRMED`: the same four-way settlement on price-action, with `PENDING` -> `WATCHING_PRICE` and `CONFIRMED` -> `READY_FOR_PLAN`. **Out-of-order use** (price-action provided while news is still `PENDING`): price-action is still fully evaluated here with the same four-way settlement (`INVALID`/`NOT_CONFIRMED`/`INSUFFICIENT` resolve normally), but a `CONFIRMED` price-action caps the status at `CROWDED` with warning `out_of_order_price_action` instead of advancing to `READY_FOR_PLAN`, since news was never actually confirmed. If both news and price-action are `PENDING` -> `CROWDED` (crowding only).
Pipeline order therefore **refines**, rather than simply restates, "any explicit NOT_CONFIRMED rejects": a later step's `NOT_CONFIRMED` can be masked by an EARLIER step's own settlement first (see the symmetric pin below) -- each later step's validity is premised on the earlier ones having already passed, mirroring how technical-analyst itself refuses to run its own confirmation pass against a `NEUTRAL` (unconfirmed) detector.
## Full State Table
C has 5 reachable labels: `CROWDED_LONG` and `CROWDED_SHORT` (both `CONFIRMED`, differing only in fade direction), `NOT_CONFIRMED`, `INSUFFICIENT`, `INVALID`. N and P each have 5 reachable labels: `CONFIRMED`, `NOT_CONFIRMED`, `INSUFFICIENT`, `PENDING`, `INVALID`. All 125 combinations are exhaustively unit-tested; the table below collapses them by which step settles the decision.
| C state | N state | P state | Result |
|---|---|---|---|
| `INVALID` or `INSUFFICIENT` | any | any | `INSUFFICIENT_EVIDENCE` (step 1 only) |
| `NOT_CONFIRMED` | any | any | `REJECTED` (step 1 only) |
| `CONFIRMED` | `INVALID` | any | `INSUFFICIENT_EVIDENCE` (step 2; P never inspected) |
| `CONFIRMED` | `NOT_CONFIRMED` | any | `REJECTED` (step 2; P never inspected) |
| `CONFIRMED` | `INSUFFICIENT` | any | `INSUFFICIENT_EVIDENCE` (step 2; P never inspected) |
| `CONFIRMED` | `PENDING` | `INVALID` | `INSUFFICIENT_EVIDENCE` (step 3, out-of-order) |
| `CONFIRMED` | `PENDING` | `NOT_CONFIRMED` | `REJECTED` (step 3, out-of-order) |
| `CONFIRMED` | `PENDING` | `INSUFFICIENT` | `INSUFFICIENT_EVIDENCE` (step 3, out-of-order) |
| `CONFIRMED` | `PENDING` | `CONFIRMED` | `CROWDED` (+ `out_of_order_price_action` warning) |
| `CONFIRMED` | `PENDING` | `PENDING` | `CROWDED` |
| `CONFIRMED` | `CONFIRMED` | `INVALID` | `INSUFFICIENT_EVIDENCE` (step 3) |
| `CONFIRMED` | `CONFIRMED` | `NOT_CONFIRMED` | `REJECTED` (step 3) |
| `CONFIRMED` | `CONFIRMED` | `INSUFFICIENT` | `INSUFFICIENT_EVIDENCE` (step 3) |
| `CONFIRMED` | `CONFIRMED` | `PENDING` | `WATCHING_PRICE` |
| `CONFIRMED` | `CONFIRMED` | `CONFIRMED` | `READY_FOR_PLAN` |
## Named Precedence Pins
These specific combinations are individually pinned in the test suite because they resolve an ambiguity a naive implementation could get wrong:
- **C=`NOT_CONFIRMED` + N=`INVALID` (unreadable)** -> `REJECTED`. Crowding's own conclusion is never softened by a corrupted downstream file.
- **C=`NOT_CONFIRMED` + N=`INVALID` (symbol_mismatch)** -> `REJECTED`. A consistency-check failure is a PER-INPUT `INVALID`, never a global override -- it cannot soften crowding's own conclusion either.
- **C=`NOT_CONFIRMED` + P=`NOT_CONFIRMED`** -> `REJECTED`, with crowding named as the rejector (not price action).
- **C=`INVALID` + N=`NOT_CONFIRMED`** -> `INSUFFICIENT_EVIDENCE` (step 1 settles before N is even inspected).
- **C=`INSUFFICIENT` + N=`NOT_CONFIRMED`** -> `INSUFFICIENT_EVIDENCE` (same).
- **C=`CONFIRMED` + N=`INVALID` + P=`NOT_CONFIRMED`** -> `INSUFFICIENT_EVIDENCE` (news settles at step 2; price is never reached).
- **C=`CONFIRMED` + N=`NOT_CONFIRMED` + P=`INSUFFICIENT`** -> `REJECTED` (news settles at step 2; price is never reached).
- **C=`CONFIRMED` + N=`NOT_CONFIRMED` + P=`INVALID`(unreadable/binary)** -> `REJECTED`, **not** `INSUFFICIENT_EVIDENCE` (v4, PR #249 P1-3 -- this is the exact combination the v3 aggregate rule got wrong: news's definitive rejection at step 2 means price-action's corruption at step 3 is never even inspected).
- **Symmetric case: C=`CONFIRMED` + N=`INSUFFICIENT` + P=`NOT_CONFIRMED`** -> `INSUFFICIENT_EVIDENCE`, **not** `REJECTED` (news settles first as INSUFFICIENT; price's NOT_CONFIRMED is never reached).
- **Out-of-order P without N, P `CONFIRMED`** -> `CROWDED` + warning.
- **Out-of-order P without N, P `NOT_CONFIRMED`** -> `REJECTED` (price-action is still fully evaluated in the out-of-order branch -- out-of-order capping only applies to a `CONFIRMED` price-action).
## Cross-Input Consistency (PER-INPUT INVALID)
A consistency failure marks the **exhibiting input** `INVALID` with a named reason and flows into the precedence rules above exactly like a loader failure -- it is never a global override.
- `symbol` in a provided news or price-action report that does not match `--symbol` -> that input `INVALID` (`news_symbol_mismatch` / `price_action_symbol_mismatch`). For the detector, `--symbol` simply not being found in `markets[]` is the existing `INSUFFICIENT` path (`detector_missing_symbol`), not a mismatch.
- `direction` in a provided news or price-action report that does not equal the detector's `classification` -> that input `INVALID` (`news_direction_mismatch` / `price_action_direction_mismatch`). Evaluated **only** when the detector row itself is usable (crowding state `CONFIRMED`) -- an unusable detector has no classification to compare against. A `direction` value of `null` is never treated as a mismatch: it is the report's own upstream fail-closed exit (e.g. NRF's `no_direction_provided`), so it normalizes to `INSUFFICIENT` with that upstream `verdict_reason` when present, or `<input>_malformed` when it is not -- comparing `null` against the detector's classification would otherwise always be unequal and misreport a legitimate upstream insufficiency as a mismatch.
- A missing `direction` key in a provided news or price-action report -> `<input>_malformed` (a required key, checked alongside `symbol`/`verdict`/`confidence`).
- `schema_version` outside major version `1` -> that input `INVALID` (`<input>_schema_unsupported`). Read locations differ per input: the detector and news reports carry `schema_version` at the **top level**; the price-action report carries it **only** at `run_context.schema_version` -- there is no top-level key on that report, so reading the top level there would silently disable the check.
- Duplicate `symbol` rows in the detector's `markets[]` -> the **first** match wins (the same `next()` pattern the detector itself uses internally).
## Reason-Token Glossary
### Crowding (cot-contrarian-detector)
| Reason | State | Meaning |
|---|---|---|
| `detector_unreadable` | INVALID | File missing, unreadable, or not valid UTF-8 |
| `detector_parse_error` | INVALID | File read but is not valid JSON |
| `detector_non_finite` | INVALID | Valid JSON, but the parsed structure contains a non-finite float (`inf`/`-inf`/`nan`) SOMEWHERE, at any depth, in any field -- including a syntactically valid JSON number that overflows to `inf` on parse (e.g. `1e309`), and the bare `Infinity`/`-Infinity`/`NaN` literals `json.loads` accepts by default as a non-standard extension. Detected by a whole-file iterative scan in the CLI's `load_json_file`, BEFORE the data is ever handed to `normalize_crowding` -- not scoped to fields the gate reads. See "Why the non-finite scan is whole-file" below (PR #249 user-review round 3) |
| `detector_malformed` | INVALID | Valid JSON but the top level is not an object |
| `detector_schema_unsupported` | INVALID | `schema_version` major is not `1` |
| `detector_missing_symbol` | INSUFFICIENT | Symbol absent from `markets[]`, or present in `skipped[]` |
| `detector_missing_data_date` | INVALID | `run_context.data_date` missing or empty |
| `detector_invalid_data_date` | INVALID | `data_date` not a string, or unparsable |
| `detector_future_data_date` | INVALID | `data_date` is after `--as-of` |
| `detector_json_stale` | INVALID | `data_date` age exceeds `--max-detector-age-days` |
| `detector_unknown_classification` | INVALID | `classification` is not a string, or is a string not one of `CROWDED_LONG` / `CROWDED_SHORT` / `NEUTRAL`. Type-checked before the allowlist membership test -- an unhashable value (a JSON list/dict) would otherwise crash that check instead of failing closed (PR #249 P1-1) |
| `detector_not_crowded` | NOT_CONFIRMED | `classification` is `NEUTRAL` -- measurably not crowded, an explicit negative |
### News Failure (news-reaction-failure-analyzer) and Price Action (technical-analyst)
Both reports share the same reason-token shape, prefixed `news_` / `price_action_` respectively:
| Reason (news / price_action) | State | Meaning |
|---|---|---|
| `_unreadable` | INVALID | File missing, unreadable, or not valid UTF-8 |
| `_parse_error` | INVALID | File read but is not valid JSON |
| `_non_finite` | INVALID | Valid JSON, but a non-finite float exists somewhere in the parsed structure -- same whole-file scan as `detector_non_finite` above, run before this report's data reaches `normalize_news` / `normalize_price_action` |
| `_malformed` | INVALID | Top level is not an object; `symbol`/`direction`/`verdict`/`confidence` is missing; `direction` is present but neither `null` nor a string; or `verdict`/`confidence` is present but not a string (a JSON list/dict/number/bool/`null` where a string is required) |
| `_symbol_mismatch` | INVALID | Report's `symbol` does not equal `--symbol` |
| `_schema_unsupported` | INVALID | `schema_version` major is not `1` |
| `_direction_mismatch` | INVALID | Report's `direction` is a non-null string that does not equal the detector's `classification` (checked only when crowding is usable). A `null` `direction` is never a mismatch -- see "Cross-Input Consistency" above. A non-string, non-null `direction` is `_malformed`, not a mismatch (PR #249 P1-1) |
| `_missing_as_of` | INVALID | `run_context.as_of` missing or empty |
| `_invalid_as_of` | INVALID | `as_of` not a string, or unparsable |
| `_future_as_of` | INVALID | `as_of` is after `--as-of` |
| `_json_stale` | INVALID | `as_of` age exceeds `--max-report-age-days` |
| `_unknown_verdict` | INVALID | `verdict` is a string but not one of the three known values for that report type. (A non-string `verdict` is `_malformed` instead, checked first -- `verdict not in valid_verdicts` requires a hashable operand, so type-checking first avoids a crash on an unhashable value like a JSON list; PR #249 P1-1, the reported repro) |
| `_unknown_confidence` | INVALID | `confidence` is a string but not one of `HIGH` / `MEDIUM` / `LOW`. Same type-check-first pattern as `_unknown_verdict` (PR #249 P1-1/P1-2 -- an unvalidated confidence used to reach `gate_confidence` in the output verbatim, e.g. `"BANANA"`). `LOW` is accepted: both upstreams document it as a reserved token they never actually emit, ranked weakest in `gate_confidence`'s weakest-link computation |
| (upstream `verdict_reason`) | NOT_CONFIRMED / INSUFFICIENT | The upstream report's own `verdict_reason` is carried through unchanged (e.g. `no_reversal_evidence`, `no_usable_events`) |
### Price-Action's CONFIRMED path: two more checks, and an asymmetry with news
`price_action_missing_stop_reference` (INVALID) fires when `verdict` is `CONFIRMED` but neither `handoff.price_action.stop_reference` nor the top-level `swing_levels.stop_reference` provided any non-null value. `price_action_invalid_stop_reference` (INVALID) fires when a source DID provide an explicit, non-null value that is unusable: not an `int`/`float`, a JSON boolean (`isinstance(True, int)` is `True` in Python, so a bare `true` must be excluded explicitly), zero, or negative (PR #249 user-review round 2, P1-B). A non-finite value (`Infinity`/`-Infinity`/`NaN`, including a syntactically valid JSON number like `1e309` that overflows to `inf` on parse) never reaches this check at all as of round 3 -- see "Why the non-finite scan is whole-file" below; it is caught earlier, at load time, as `price_action_non_finite`. An explicit-but-invalid value in `handoff.price_action.stop_reference` does **not** fall back to `swing_levels.stop_reference` -- an explicit garbage value is a real report bug, not an absence to paper over; the fallback is for when `handoff` provides no value at all.
`price_action_unknown_reason` (INVALID) and a stricter `price_action_malformed` fire on the CONFIRMED path's `verdict_reason`: a non-string or empty value is `price_action_malformed`; a well-typed string outside `{key_reversal, failed_extreme, failed_breakout}` (technical-analyst's `weekly_price_action.CHECK_REASON_MAP` values -- pinned by a cross-skill consistency test that imports TA's own source, so upstream drift there is caught, not silently trusted) is `price_action_unknown_reason` (PR #249 user-review round 2, P1-A). This is an intentional **asymmetry with news**: news's `verdict_reason` stays type-checked-but-display-only (a non-string value degrades to a generic fallback token, never rejects the input) because NRF's reason vocabulary is open-ended and the gate never *acts* on it. Price-action's CONFIRMED `verdict_reason` is different -- it feeds directly into `entry_trigger`, part of the actionable `READY_FOR_PLAN` output -- so it is allowlisted against TA's actual contract rather than merely type-checked.
**`READY_FOR_PLAN` invariant:** whenever `setup_status` is `READY_FOR_PLAN`, `entry_trigger` is guaranteed to be a non-empty string and `invalidation_level` is guaranteed to be a finite, positive number. This is enforced twice: primarily by `normalize_price_action`'s CONFIRMED-path validation above (which can never return `CONFIRMED` without both being valid), and defensively by an assertion inside `build_gate_result` itself, as a second independent guard against a future regression.
### Why the non-finite scan is whole-file, not field-scoped
`gate_logic.normalize_*` intentionally echoes some raw-but-invalid values verbatim into a `NormalizedInput` for audit transparency -- e.g. an unknown `classification` is still stored so the written report shows exactly what bad value was present. The CLI's JSON writer also uses `json.dumps(..., allow_nan=False)` as a defense against ever emitting a non-standard `Infinity`/`-Infinity`/`NaN` token. Those two design choices collided in a real regression (PR #249 user-review round 3): `classification: 1e309` parses to Python's `inf`, normalizes correctly to `detector_unknown_classification` -- but that INVALID `NormalizedInput` still echoed the raw `inf` value, and writing the report then crashed on `allow_nan=False` with `ValueError`, breaking the exit-0-always-writes-a-report contract on a code path entirely different from the one that determined the input was bad.
The fix is a whole-file scan (`run_contrarian_setup_gate.py`'s `_contains_non_finite`, called from `load_json_file` immediately after `json.loads`) that rejects a report if a non-finite float exists ANYWHERE in it -- even in a field the gate's decision logic never reads (a `skipped[]` entry's incidental note, a deeply-nested `run_context` field, ...). This is deliberately coarser than validating only the specific fields the gate consumes: it guarantees no raw non-finite value can ever reach `gate_logic` at all, so no future echo site anywhere in the module needs to remember to individually guard against this -- the loader closes the class of bug structurally, once, rather than requiring every current and future audit-echo field to defend itself. `allow_nan=False` remains in place as a second, now-normally-unreachable defense layer.
**`_contains_non_finite` is iterative, not recursive** (PR #249 user-review round 4): a legitimate, ordinary JSON document can be nested hundreds of levels deep -- an array-of-arrays field the gate never reads, say -- and is perfectly valid input, not an attack. A recursive walker over that structure raises `RecursionError` well before Python's default limit would suggest (confirmed empirically: each JSON nesting level costs more than one Python call frame once the `any(... for ...)` generator-expression overhead is counted, so a plain recursive version of this function starts failing around depth ~500), which used to crash the CLI with exit 1 on a file that should have produced a completely ordinary result. The iterative version uses an explicit stack instead -- no call-stack depth limit, bounded only by memory, O(n) over the structure.
`json.loads` itself can also raise `RecursionError` on sufficiently extreme nesting, independent of this module's own scan -- confirmed empirically in this environment: the C-accelerated decoder handles depth 100,000 without issue but raises `RecursionError` ("Stack overflow ... while decoding a JSON array") around depth 200,000. `RecursionError` is not a subclass of `json.JSONDecodeError` or `ValueError`, so `load_json_file` catches it explicitly and routes it to the same `<input>_parse_error` class as any other unparsable file -- exit 0, a report is still written, never an uncaught crash.
## Warnings (Never Change the Status)
- `price_action_confidence_medium` / `news_confidence_medium` -- fires only at `READY_FOR_PLAN` when that input's confidence is `MEDIUM` (single-signal weakness).
- `<input>_near_stale` -- fires when an input's age is within 2 days of its configured max age (and not already over it, which would instead be `INVALID`/`_json_stale`).
- `detector_data_date_divergence` -- the crowding market row's own `data_date` differs from `run_context.data_date` (the run's vintage, which is what staleness is evaluated against).
- `out_of_order_price_action` -- see rule 5 above.
## Worked Example: Real B6 REJECTED Case
Regenerated live against `cot-contrarian-detector` on 2026-07-12 (`--symbols B6,BT,D6 --as-of 2026-07-12`): B6 (British Pound) came back `CROWDED_SHORT` with `cot_index_3y=7.2`, `run_context.data_date=2026-07-07`. Running the gate with `--as-of 2026-07-15` (detector age 8 days, under the default 10-day max) and a news-reaction-failure-analyzer report whose `verdict` is `NOT_CONFIRMED`:
- Crowding normalizes to `CONFIRMED`, `classification=CROWDED_SHORT`, `direction=LONG` (step 1 does not stop the pipeline -- crowding is usable).
- News normalizes to `NOT_CONFIRMED` and settles the decision at step 2 -- price-action is never inspected at all for this run (there isn't one provided in this example anyway).
- `setup_status = REJECTED`, `missing_confirmations = [{"step": "news_failure", "state": "NOT_CONFIRMED", "reason": "<the news report's own verdict_reason>"}]`.
- `direction` remains `LONG` in the output (crowding was confirmed, even though the overall setup was rejected) -- an audit trail of what crowd was being faded, not an actionable signal.
Running the same detector report with `--as-of` set to a later date pushes the detector's age past `--max-detector-age-days`, which instead produces `INSUFFICIENT_EVIDENCE` with reason `detector_json_stale` -- crowding itself becomes unusable at step 1, before news is ever inspected.
requirements.txt
# stdlib-only: packaged scripts use the Python standard library only.
scripts/gate_logic.py
#!/usr/bin/env python3
"""
Contrarian Setup Gate -- pure synthesis logic.
The center of Jason Shapiro's contrarian pipeline: this module normalizes
the three upstream verdicts (crowding / news-failure / price-action) into
one fail-closed `NormalizedInput` per step, then combines them through an
explicit, exhaustively-tested precedence state machine into one actionable
`setup_status`.
This module is PURE: no file I/O, no network, no environment reads. The CLI
(`run_contrarian_setup_gate.py`) owns file loading -- unreadable /
parse_error / non_finite are CLI-level `load_error` tags detected before
this module ever sees the data (non_finite: PR #249 user-review round 3,
a whole-file recursive scan for any non-finite float anywhere, closing off
the only route by which a raw non-finite value could otherwise reach the
CLI's `allow_nan=False` JSON writer via an audit-echo field) -- and passes
either the parsed JSON or one of those tags into the `normalize_*`
functions below, which detect the remaining classes (malformed / stale /
symbol_mismatch / direction_mismatch / schema_unsupported / unknown_verdict
/ unknown_confidence / unknown_reason / invalid_stop_reference) themselves.
State machine precedence (plan Issue #241 §3.3, v4 -- SEQUENTIAL
pipeline-order evaluation, PR #249 user-review P1-3):
Each step is evaluated in strict pipeline order -- crowding, then news,
then price-action -- and each step FULLY SETTLES (reaches a
determination) before the next step's file is even consulted for a
decision. This is deliberate, not an optimization: an earlier step's
definitive verdict must never be softened by a LATER step's problem
(corruption, staleness, its own insufficiency, ...). The v3 design
aggregated "any provided INVALID" and "any NOT_CONFIRMED" ACROSS both
downstream steps before deciding which rule fired first; that let a
later step's file corruption soften an earlier step's definitive
NOT_CONFIRMED rejection back up to a mere INSUFFICIENT_EVIDENCE -- the
same class of softening bug already fixed twice elsewhere in this gate
(the loader path and the consistency path), now caught a third time in
cross-step aggregation by an independent user review of PR #249.
1. Crowding: INVALID/INSUFFICIENT -> INSUFFICIENT_EVIDENCE;
NOT_CONFIRMED (classification NEUTRAL) -> REJECTED; CONFIRMED ->
continue to step 2. (Crowding has no PENDING state -- the detector
report is always required.)
2. News, if provided (state != PENDING): INVALID -> INSUFFICIENT_EVIDENCE;
NOT_CONFIRMED -> REJECTED; INSUFFICIENT -> INSUFFICIENT_EVIDENCE;
CONFIRMED -> continue to step 3. If news is PENDING, price-action is
still evaluated next under the out-of-order rule (step 3).
3. Price-action, if provided: the same four-way settlement as step 2.
Out-of-order use (price-action provided while news is still PENDING):
price-action is STILL fully evaluated here -- INVALID/NOT_CONFIRMED/
INSUFFICIENT still resolve to their normal outcomes -- but a CONFIRMED
price-action caps the final status at CROWDED with warning
`out_of_order_price_action` rather than advancing to READY_FOR_PLAN,
since news was never actually confirmed.
4. Progressive from the confirmed prefix: crowding only -> CROWDED;
crowding+news -> WATCHING_PRICE; crowding+news+price-action ->
READY_FOR_PLAN (direction, gate_confidence, entry_trigger,
invalidation_level populated).
Pipeline order therefore REFINES the issue's original "any explicit
NOT_CONFIRMED rejects" framing: a later step's NOT_CONFIRMED can still
be masked by an EARLIER step's own settlement (e.g. news=INSUFFICIENT
settles before price-action=NOT_CONFIRMED is ever looked at, giving
INSUFFICIENT_EVIDENCE, not REJECTED) -- each later step's validity is
premised on the earlier ones already having passed, mirroring how
technical-analyst itself refuses to run its own confirmation pass on a
NEUTRAL (unconfirmed) detector.
Cross-input consistency (symbol_mismatch / direction_mismatch /
schema_unsupported) is PER-INPUT INVALID (v3): a failure marks the
exhibiting input INVALID with a named reason and flows into the rules
above exactly like a loader failure -- never a global override that could
soften crowding's own conclusion.
"""
from __future__ import annotations
import math
from dataclasses import dataclass
from datetime import date, datetime
from typing import Any
SCHEMA_VERSION = "1.0"
SKILL_NAME = "contrarian-setup-gate"
STATE_CONFIRMED = "CONFIRMED"
STATE_NOT_CONFIRMED = "NOT_CONFIRMED"
STATE_INSUFFICIENT = "INSUFFICIENT"
STATE_PENDING = "PENDING"
STATE_INVALID = "INVALID"
STEP_CROWDING = "crowding"
STEP_NEWS = "news_failure"
STEP_PRICE = "price_action"
# CROWDED_LONG means the crowd is long -> fade SHORT. CROWDED_SHORT means
# the crowd is short -> fade LONG. NEUTRAL is not one of these keys (it
# maps to NOT_CONFIRMED, not a fade direction).
FADE_DIRECTION = {"CROWDED_LONG": "SHORT", "CROWDED_SHORT": "LONG"}
NEWS_VERDICTS = {"CONFIRMED", "NOT_CONFIRMED", "INSUFFICIENT_EVIDENCE"}
PRICE_VERDICTS = {"CONFIRMED", "NOT_CONFIRMED", "INSUFFICIENT_DATA"}
# LOW is a reserved token both upstreams document but never actually emit
# (technical-analyst's compute_confidence(): "LOW is reserved, never
# emitted"). Accepted here for forward-compatibility, ranked weakest.
# Anything else (non-string, or a string outside this set) is INVALID
# `<input>_unknown_confidence` -- never silently passed through (PR #249
# user-review P1-2: an unvalidated "BANANA" used to reach gate_confidence
# verbatim).
VALID_CONFIDENCES = {"HIGH", "MEDIUM", "LOW"}
CONFIDENCE_RANK = {"HIGH": 3, "MEDIUM": 2, "LOW": 1}
# "Near stale" fires when age is within this many days of the configured
# max age (and not already over it, which is INVALID `<input>_json_stale`).
NEAR_STALE_WINDOW_DAYS = 2
# Maps TA's short verdict_reason token (plan §2 output contract) back to
# the `checks` dict key that carries its `week_of`, for entry_trigger.
CHECK_REASON_TO_KEY = {
"key_reversal": "weekly_key_reversal",
"failed_extreme": "failed_extreme",
"failed_breakout": "failed_breakout",
}
# The exhaustive set of verdict_reason tokens technical-analyst can emit
# for a CONFIRMED verdict (weekly_price_action.CHECK_REASON_MAP's values,
# verified against the merged source). Derived from CHECK_REASON_TO_KEY's
# keys so the two can never drift apart within this module; a cross-skill
# consistency test in test_gate_logic.py additionally pins this against
# TA's own source, so upstream drift there is caught, not silently
# trusted (PR #249 user-review round 2, P1-A). Unlike news's verdict_reason
# (display-only, open-ended vocabulary, never allowlisted -- see
# references/gate-decision-table.md), price-action's CONFIRMED
# verdict_reason feeds directly into `entry_trigger`, part of the
# actionable READY_FOR_PLAN output -- so it IS allowlisted here.
PRICE_ACTION_CONFIRMED_REASONS = frozenset(CHECK_REASON_TO_KEY)
REQUIRED_REPORT_KEYS = ("symbol", "direction", "verdict", "confidence")
@dataclass(frozen=True)
class NormalizedInput:
"""One step's normalized, fail-closed state.
`kind` selects which fields `to_audit_dict()` surfaces, matching the
`inputs.<step>` block shape in the output contract (plan §2). `reason`
is populated for INVALID (a gate-side validation failure, named token)
and for NOT_CONFIRMED/INSUFFICIENT derived from an upstream verdict
(the upstream's own `verdict_reason`, when present -- more informative
than a generic token).
"""
kind: str
state: str
reason: str | None = None
warnings: tuple[str, ...] = ()
classification: str | None = None
direction: str | None = None # crowding only: fade direction SHORT/LONG
verdict: str | None = None
confidence: str | None = None
verdict_reason: str | None = None
stop_reference: float | None = None
entry_trigger: str | None = None
data_date: str | None = None
as_of: str | None = None
age_days: int | None = None
report_path: str | None = None
def to_audit_dict(self) -> dict[str, Any]:
if self.kind == STEP_CROWDING:
return {
"state": self.state,
"classification": self.classification,
"data_date": self.data_date,
"age_days": self.age_days,
"report_path": self.report_path,
}
if self.kind == STEP_NEWS:
return {
"state": self.state,
"verdict": self.verdict,
"confidence": self.confidence,
"verdict_reason": self.verdict_reason,
"as_of": self.as_of,
"age_days": self.age_days,
"report_path": self.report_path,
}
return { # STEP_PRICE
"state": self.state,
"verdict": self.verdict,
"confidence": self.confidence,
"verdict_reason": self.verdict_reason,
"stop_reference": self.stop_reference,
"as_of": self.as_of,
"age_days": self.age_days,
"report_path": self.report_path,
}
def pending_input(kind: str) -> NormalizedInput:
"""The normalized state for a step whose report file was not provided."""
return NormalizedInput(kind=kind, state=STATE_PENDING)
# --- Shared helpers --------------------------------------------------------
def _is_supported_schema(value: Any) -> bool:
"""Accept schema major version "1" (e.g. "1.0", "1.4"); reject anything
else, including missing/non-string/other-major values (fail closed)."""
if not isinstance(value, str) or not value:
return False
return value.split(".", 1)[0] == "1"
def _parse_as_of_date(as_of: str) -> date:
return datetime.strptime(as_of, "%Y-%m-%d").date()
def _resolve_vintage(raw_value: Any, as_of_dt: date, max_age_days: int) -> tuple[str, int | None]:
"""Resolve a report's vintage string (detector `data_date` or
news/price `run_context.as_of`) against `as_of_dt`.
Returns (status, age_days) where status is one of "missing" (None/""),
"invalid" (not a string, or unparsable), "future" (dated after
as_of_dt), "stale" (older than max_age_days), or "ok". Mirrors the
proven guard sequence in technical-analyst's
resolve_direction_from_detector() and news-reaction-failure-analyzer's
counterpart: type guard before slicing, no as_of-for-data_date
fallback, age computed only once the value is known-good.
"""
if raw_value is None or raw_value == "":
return "missing", None
if not isinstance(raw_value, str):
return "invalid", None
try:
value_dt = datetime.strptime(raw_value[:10], "%Y-%m-%d").date()
except ValueError:
return "invalid", None
age_days = (as_of_dt - value_dt).days
if age_days < 0:
return "future", age_days
if age_days > max_age_days:
return "stale", age_days
return "ok", age_days
def _near_stale(age_days: int, max_age_days: int) -> bool:
return (max_age_days - age_days) <= NEAR_STALE_WINDOW_DAYS
def _valid_stop_reference(value: Any) -> float | None:
"""A usable stop_reference: a finite, positive, non-bool number.
`isinstance(True, int)` is True in Python, so bool must be excluded
explicitly, or a boolean would silently pass as 0.0/1.0. Non-finite
values (inf/-inf/NaN -- all of which `json.loads` accepts as literal
tokens even though they aren't standard JSON) are rejected here too:
a non-finite stop_reference used to flow straight through to
READY_FOR_PLAN's `invalidation_level` and produce non-standard JSON
output (PR #249 user-review round 2, P1-B)."""
if isinstance(value, bool) or not isinstance(value, (int, float)):
return None
if not math.isfinite(value) or value <= 0:
return None
return float(value)
def _resolve_stop_reference(raw_data: dict[str, Any]) -> tuple[float | None, str | None]:
"""handoff.price_action.stop_reference, falling back to top-level
swing_levels.stop_reference when handoff provides no value at all
(plan §3.1). Returns (value, error): error is `None` on success,
`"missing"` when neither source provided any non-null value, or
`"invalid"` when a source provided an explicit but unusable value
(wrong type, bool, non-finite, zero, or negative) -- an explicit
garbage value is a real report bug and must not be silently masked
by falling through to the other source."""
handoff = raw_data.get("handoff")
if isinstance(handoff, dict):
price_action = handoff.get("price_action")
if isinstance(price_action, dict):
raw_value = price_action.get("stop_reference")
if raw_value is not None:
value = _valid_stop_reference(raw_value)
return (value, None) if value is not None else (None, "invalid")
swing_levels = raw_data.get("swing_levels")
if isinstance(swing_levels, dict):
raw_value = swing_levels.get("stop_reference")
if raw_value is not None:
value = _valid_stop_reference(raw_value)
return (value, None) if value is not None else (None, "invalid")
return None, "missing"
def _build_entry_trigger(raw_data: dict[str, Any], verdict_reason: Any) -> str | None:
"""Factual echo of the TA confirming signal: `verdict_reason` plus the
matching check's `week_of`, when both are resolvable. Never raises on
an unexpected shape -- degrades to a shorter echo, or None."""
if not isinstance(verdict_reason, str) or not verdict_reason:
return None
week_of = None
checks = raw_data.get("checks")
if isinstance(checks, dict):
check_key = CHECK_REASON_TO_KEY.get(verdict_reason)
if check_key:
check = checks.get(check_key)
if isinstance(check, dict):
candidate = check.get("week_of")
if isinstance(candidate, str) and candidate:
week_of = candidate
if week_of:
return f"price-action confirmation: {verdict_reason} at week_of={week_of}"
return f"price-action confirmation: {verdict_reason}"
# --- Per-step normalization -------------------------------------------------
def normalize_crowding(
raw_data: Any,
load_error: str | None,
*,
symbol: str,
as_of: str,
max_age_days: int,
report_path: str | None = None,
) -> NormalizedInput:
"""Normalize a cot-contrarian-detector report for `symbol`.
CROWDED_LONG/CROWDED_SHORT -> CONFIRMED (direction = fade side).
NEUTRAL -> NOT_CONFIRMED (measurably not crowded, an explicit
negative -- not an insufficiency). Symbol absent from markets[] or
present in skipped[] -> INSUFFICIENT (detector_missing_symbol).
Everything else fails closed to INVALID with a named reason.
"""
kind = STEP_CROWDING
if load_error == "unreadable":
return NormalizedInput(
kind=kind, state=STATE_INVALID, reason="detector_unreadable", report_path=report_path
)
if load_error == "parse_error":
return NormalizedInput(
kind=kind, state=STATE_INVALID, reason="detector_parse_error", report_path=report_path
)
if load_error == "non_finite":
return NormalizedInput(
kind=kind, state=STATE_INVALID, reason="detector_non_finite", report_path=report_path
)
if not isinstance(raw_data, dict):
return NormalizedInput(
kind=kind, state=STATE_INVALID, reason="detector_malformed", report_path=report_path
)
if not _is_supported_schema(raw_data.get("schema_version")):
return NormalizedInput(
kind=kind,
state=STATE_INVALID,
reason="detector_schema_unsupported",
report_path=report_path,
)
run_context = raw_data.get("run_context")
run_context = run_context if isinstance(run_context, dict) else {}
markets_raw = raw_data.get("markets")
markets = markets_raw if isinstance(markets_raw, list) else []
skipped_raw = raw_data.get("skipped")
skipped_list = skipped_raw if isinstance(skipped_raw, list) else []
# Only string symbols are ever eligible to match --symbol (itself a
# string); filtering here also avoids ever trying to add an unhashable
# value (a list/dict "symbol" in an untrusted skip-entry) to this set,
# which would crash the set comprehension itself (PR #249 P1-1).
skipped_symbols = {
s.get("symbol")
for s in skipped_list
if isinstance(s, dict) and isinstance(s.get("symbol"), str)
}
# Duplicate symbol rows in markets[] -> first match wins (the `next()`
# pattern the detector itself uses; plan §3.2 v2 P2-7).
market_row = next(
(m for m in markets if isinstance(m, dict) and m.get("symbol") == symbol), None
)
if market_row is None or symbol in skipped_symbols:
return NormalizedInput(
kind=kind,
state=STATE_INSUFFICIENT,
reason="detector_missing_symbol",
report_path=report_path,
)
data_date = run_context.get("data_date")
as_of_dt = _parse_as_of_date(as_of)
status, age_days = _resolve_vintage(data_date, as_of_dt, max_age_days)
if status == "missing":
return NormalizedInput(
kind=kind,
state=STATE_INVALID,
reason="detector_missing_data_date",
report_path=report_path,
)
if status == "invalid":
return NormalizedInput(
kind=kind,
state=STATE_INVALID,
reason="detector_invalid_data_date",
report_path=report_path,
)
if status == "future":
return NormalizedInput(
kind=kind,
state=STATE_INVALID,
reason="detector_future_data_date",
data_date=data_date,
age_days=age_days,
report_path=report_path,
)
if status == "stale":
return NormalizedInput(
kind=kind,
state=STATE_INVALID,
reason="detector_json_stale",
data_date=data_date,
age_days=age_days,
report_path=report_path,
)
assert age_days is not None # status == "ok" always sets age_days
warnings: list[str] = []
if _near_stale(age_days, max_age_days):
warnings.append("detector_near_stale")
row_data_date = market_row.get("data_date")
if isinstance(row_data_date, str) and row_data_date and row_data_date != data_date:
warnings.append("detector_data_date_divergence")
classification = market_row.get("classification")
if classification == "NEUTRAL":
return NormalizedInput(
kind=kind,
state=STATE_NOT_CONFIRMED,
reason="detector_not_crowded",
classification=classification,
data_date=data_date,
age_days=age_days,
warnings=tuple(warnings),
report_path=report_path,
)
# Type guard BEFORE the allowlist membership check: `in`/`not in` on a
# dict requires the operand to be hashable, so an unhashable
# classification (e.g. a JSON list) would otherwise crash this line
# with TypeError instead of failing closed (PR #249 P1-1).
if not isinstance(classification, str) or classification not in FADE_DIRECTION:
return NormalizedInput(
kind=kind,
state=STATE_INVALID,
reason="detector_unknown_classification",
classification=classification,
data_date=data_date,
age_days=age_days,
warnings=tuple(warnings),
report_path=report_path,
)
return NormalizedInput(
kind=kind,
state=STATE_CONFIRMED,
classification=classification,
direction=FADE_DIRECTION[classification],
data_date=data_date,
age_days=age_days,
warnings=tuple(warnings),
report_path=report_path,
)
def _normalize_downstream_report(
raw_data: Any,
load_error: str | None,
*,
kind: str,
symbol: str,
as_of: str,
max_age_days: int,
detector: NormalizedInput,
report_path: str | None,
valid_verdicts: set[str],
verdict_to_state: dict[str, str],
prefix: str,
) -> NormalizedInput:
"""Shared normalization body for news-failure and price-action reports
(both share the same top-level shape: symbol/direction/verdict/
confidence/verdict_reason/run_context.as_of). Only price-action needs
the extra stop_reference/entry_trigger handling, layered on by its own
wrapper below."""
if load_error == "unreadable":
return NormalizedInput(
kind=kind, state=STATE_INVALID, reason=f"{prefix}_unreadable", report_path=report_path
)
if load_error == "parse_error":
return NormalizedInput(
kind=kind, state=STATE_INVALID, reason=f"{prefix}_parse_error", report_path=report_path
)
if load_error == "non_finite":
return NormalizedInput(
kind=kind, state=STATE_INVALID, reason=f"{prefix}_non_finite", report_path=report_path
)
if not isinstance(raw_data, dict):
return NormalizedInput(
kind=kind, state=STATE_INVALID, reason=f"{prefix}_malformed", report_path=report_path
)
if any(key not in raw_data for key in REQUIRED_REPORT_KEYS):
return NormalizedInput(
kind=kind, state=STATE_INVALID, reason=f"{prefix}_malformed", report_path=report_path
)
report_symbol = raw_data.get("symbol")
if (
not isinstance(report_symbol, str)
or report_symbol.strip().upper() != symbol.strip().upper()
):
return NormalizedInput(
kind=kind,
state=STATE_INVALID,
reason=f"{prefix}_symbol_mismatch",
report_path=report_path,
)
return _validated_report_input(
raw_data,
kind=kind,
as_of=as_of,
max_age_days=max_age_days,
detector=detector,
report_path=report_path,
valid_verdicts=valid_verdicts,
verdict_to_state=verdict_to_state,
prefix=prefix,
)
def _validated_report_input(
raw_data: dict[str, Any],
*,
kind: str,
as_of: str,
max_age_days: int,
detector: NormalizedInput,
report_path: str | None,
valid_verdicts: set[str],
verdict_to_state: dict[str, str],
prefix: str,
) -> NormalizedInput:
run_context = raw_data.get("run_context")
run_context = run_context if isinstance(run_context, dict) else {}
schema_source = run_context if prefix == "price_action" else raw_data
if not _is_supported_schema(schema_source.get("schema_version")):
return NormalizedInput(
kind=kind,
state=STATE_INVALID,
reason=f"{prefix}_schema_unsupported",
report_path=report_path,
)
report_direction = raw_data.get("direction")
if report_direction is None:
# The report's own direction resolution failed upstream (e.g.
# NRF's no_direction_provided early exit) -- this is that input's
# own fail-closed insufficiency, not a mismatch against the
# detector. Comparing None != detector.classification below would
# otherwise always be truthy and misreport this as
# `<prefix>_direction_mismatch` (P3-1).
fallback_reason = raw_data.get("verdict_reason")
if isinstance(fallback_reason, str) and fallback_reason:
return NormalizedInput(
kind=kind,
state=STATE_INSUFFICIENT,
reason=fallback_reason,
verdict=raw_data.get("verdict"),
confidence=raw_data.get("confidence"),
verdict_reason=fallback_reason,
report_path=report_path,
)
return NormalizedInput(
kind=kind, state=STATE_INVALID, reason=f"{prefix}_malformed", report_path=report_path
)
if not isinstance(report_direction, str):
# A non-null, non-string direction (a JSON list/dict/number/bool)
# is malformed input, not a legitimate mismatch to report against
# the detector's classification (PR #249 P1-1: every enum-shaped
# field must be type-validated before it participates in any
# comparison or membership check).
return NormalizedInput(
kind=kind, state=STATE_INVALID, reason=f"{prefix}_malformed", report_path=report_path
)
if detector.state == STATE_CONFIRMED and report_direction != detector.classification:
return NormalizedInput(
kind=kind,
state=STATE_INVALID,
reason=f"{prefix}_direction_mismatch",
report_path=report_path,
)
as_of_value = run_context.get("as_of")
as_of_dt = _parse_as_of_date(as_of)
status, age_days = _resolve_vintage(as_of_value, as_of_dt, max_age_days)
if status == "missing":
return NormalizedInput(
kind=kind,
state=STATE_INVALID,
reason=f"{prefix}_missing_as_of",
report_path=report_path,
)
if status == "invalid":
return NormalizedInput(
kind=kind,
state=STATE_INVALID,
reason=f"{prefix}_invalid_as_of",
report_path=report_path,
)
if status == "future":
return NormalizedInput(
kind=kind,
state=STATE_INVALID,
reason=f"{prefix}_future_as_of",
as_of=as_of_value,
age_days=age_days,
report_path=report_path,
)
if status == "stale":
return NormalizedInput(
kind=kind,
state=STATE_INVALID,
reason=f"{prefix}_json_stale",
as_of=as_of_value,
age_days=age_days,
report_path=report_path,
)
assert age_days is not None
warnings: list[str] = []
if _near_stale(age_days, max_age_days):
warnings.append(f"{prefix}_near_stale")
verdict = raw_data.get("verdict")
verdict_reason = raw_data.get("verdict_reason")
confidence = raw_data.get("confidence")
# Type guard BEFORE the allowlist membership check: `verdict not in
# valid_verdicts` requires verdict to be hashable, so an unhashable
# verdict (e.g. a JSON list) would otherwise crash with TypeError
# instead of failing closed (PR #249 P1-1, the reported repro). A
# non-string verdict is malformed input; a string outside the
# allowlist is an unknown (but well-typed) verdict -- two distinct,
# named reasons.
if not isinstance(verdict, str):
return NormalizedInput(
kind=kind,
state=STATE_INVALID,
reason=f"{prefix}_malformed",
verdict=verdict,
as_of=as_of_value,
age_days=age_days,
report_path=report_path,
)
if verdict not in valid_verdicts:
return NormalizedInput(
kind=kind,
state=STATE_INVALID,
reason=f"{prefix}_unknown_verdict",
verdict=verdict,
as_of=as_of_value,
age_days=age_days,
report_path=report_path,
)
# Same two-step guard for confidence (PR #249 P1-1/P1-2): an unhashable
# confidence would otherwise crash `CONFIDENCE_RANK.get(...)` deep
# inside gate_confidence computation, and an unvalidated string (e.g.
# "BANANA") would otherwise pass straight through to the output
# contract's HIGH|MEDIUM|LOW|null field.
if not isinstance(confidence, str):
return NormalizedInput(
kind=kind,
state=STATE_INVALID,
reason=f"{prefix}_malformed",
verdict=verdict,
as_of=as_of_value,
age_days=age_days,
report_path=report_path,
)
if confidence not in VALID_CONFIDENCES:
return NormalizedInput(
kind=kind,
state=STATE_INVALID,
reason=f"{prefix}_unknown_confidence",
verdict=verdict,
confidence=confidence,
as_of=as_of_value,
age_days=age_days,
report_path=report_path,
)
state = verdict_to_state[verdict]
if state == STATE_CONFIRMED and prefix == "price_action":
# verdict_reason feeds entry_trigger, part of the actionable
# READY_FOR_PLAN output -- unlike the display-only fallback below
# (used for NOT_CONFIRMED/INSUFFICIENT and for news generally),
# it is validated type-first, then against TA's own contract
# (PR #249 user-review round 2, P1-A).
if not isinstance(verdict_reason, str) or not verdict_reason:
return NormalizedInput(
kind=kind,
state=STATE_INVALID,
reason="price_action_malformed",
verdict=verdict,
confidence=confidence,
as_of=as_of_value,
age_days=age_days,
warnings=tuple(warnings),
report_path=report_path,
)
if verdict_reason not in PRICE_ACTION_CONFIRMED_REASONS:
return NormalizedInput(
kind=kind,
state=STATE_INVALID,
reason="price_action_unknown_reason",
verdict=verdict,
confidence=confidence,
verdict_reason=verdict_reason,
as_of=as_of_value,
age_days=age_days,
warnings=tuple(warnings),
report_path=report_path,
)
stop_reference, stop_error = _resolve_stop_reference(raw_data)
if stop_error is not None:
reason = (
"price_action_missing_stop_reference"
if stop_error == "missing"
else "price_action_invalid_stop_reference"
)
return NormalizedInput(
kind=kind,
state=STATE_INVALID,
reason=reason,
verdict=verdict,
confidence=confidence,
verdict_reason=verdict_reason,
as_of=as_of_value,
age_days=age_days,
warnings=tuple(warnings),
report_path=report_path,
)
return NormalizedInput(
kind=kind,
state=STATE_CONFIRMED,
verdict=verdict,
confidence=confidence,
verdict_reason=verdict_reason,
stop_reference=stop_reference,
entry_trigger=_build_entry_trigger(raw_data, verdict_reason),
as_of=as_of_value,
age_days=age_days,
warnings=tuple(warnings),
report_path=report_path,
)
reason = (
verdict_reason
if isinstance(verdict_reason, str) and verdict_reason
else f"{prefix}_{state.lower()}"
)
return NormalizedInput(
kind=kind,
state=state,
reason=reason if state != STATE_CONFIRMED else None,
verdict=verdict,
confidence=confidence,
verdict_reason=verdict_reason,
as_of=as_of_value,
age_days=age_days,
warnings=tuple(warnings),
report_path=report_path,
)
def normalize_news(
raw_data: Any,
load_error: str | None,
*,
symbol: str,
as_of: str,
max_age_days: int,
detector: NormalizedInput,
report_path: str | None = None,
) -> NormalizedInput:
"""Normalize a news-reaction-failure-analyzer report for `symbol`."""
return _normalize_downstream_report(
raw_data,
load_error,
kind=STEP_NEWS,
symbol=symbol,
as_of=as_of,
max_age_days=max_age_days,
detector=detector,
report_path=report_path,
valid_verdicts=NEWS_VERDICTS,
verdict_to_state={
"CONFIRMED": STATE_CONFIRMED,
"NOT_CONFIRMED": STATE_NOT_CONFIRMED,
"INSUFFICIENT_EVIDENCE": STATE_INSUFFICIENT,
},
prefix="news",
)
def normalize_price_action(
raw_data: Any,
load_error: str | None,
*,
symbol: str,
as_of: str,
max_age_days: int,
detector: NormalizedInput,
report_path: str | None = None,
) -> NormalizedInput:
"""Normalize a technical-analyst contrarian-confirmation report for
`symbol`. schema_version is read from `run_context.schema_version`
(TA carries no top-level key -- reading top-level here would silently
disable the check; plan §3.2 v2 P1-3)."""
return _normalize_downstream_report(
raw_data,
load_error,
kind=STEP_PRICE,
symbol=symbol,
as_of=as_of,
max_age_days=max_age_days,
detector=detector,
report_path=report_path,
valid_verdicts=PRICE_VERDICTS,
verdict_to_state={
"CONFIRMED": STATE_CONFIRMED,
"NOT_CONFIRMED": STATE_NOT_CONFIRMED,
"INSUFFICIENT_DATA": STATE_INSUFFICIENT,
},
prefix="price_action",
)
# --- State machine -----------------------------------------------------
def decide_setup_status(
crowding: NormalizedInput, news: NormalizedInput, price: NormalizedInput
) -> tuple[str, list[dict[str, Any]], list[str]]:
"""Apply the SEQUENTIAL precedence rules (module docstring, v4) to the
three normalized inputs. Returns (setup_status, missing_confirmations,
extra_warnings) -- `extra_warnings` holds machine-level warnings (only
`out_of_order_price_action` today); per-input warnings are merged by
the caller (`build_gate_result`).
Each step fully settles -- reaches a determination -- before the next
step's state is even inspected. This is what makes the "earlier
definitive verdicts are never softened by later problems" property
structural rather than something a rule-ordering audit has to keep
re-proving (PR #249 P1-3)."""
missing: list[dict[str, Any]] = []
extra_warnings: list[str] = []
# Step 1: crowding, evaluated first and exclusively.
if crowding.state in (STATE_INVALID, STATE_INSUFFICIENT):
missing.append({"step": STEP_CROWDING, "state": crowding.state, "reason": crowding.reason})
return "INSUFFICIENT_EVIDENCE", missing, extra_warnings
if crowding.state == STATE_NOT_CONFIRMED:
missing.append(
{
"step": STEP_CROWDING,
"state": STATE_NOT_CONFIRMED,
"reason": crowding.reason or "detector_not_crowded",
}
)
return "REJECTED", missing, extra_warnings
# Crowding is CONFIRMED from here.
def _settle(step: str, inp: NormalizedInput) -> tuple[str, list[dict[str, Any]]] | None:
"""Settle one downstream step (news or price-action) against its
own four-way outcome. Returns (status, missing_entries) when the
step is definitive, or None when it's CONFIRMED (continue to the
next step) -- PENDING is handled separately by each call site
since its meaning differs between news (step 2) and price-action
(step 3, where it never blocks)."""
if inp.state == STATE_INVALID:
return "INSUFFICIENT_EVIDENCE", [
{"step": step, "state": inp.state, "reason": inp.reason}
]
if inp.state == STATE_NOT_CONFIRMED:
return "REJECTED", [{"step": step, "state": inp.state, "reason": inp.reason}]
if inp.state == STATE_INSUFFICIENT:
return "INSUFFICIENT_EVIDENCE", [
{"step": step, "state": inp.state, "reason": inp.reason}
]
return None
# Step 2: news. Fully settles before price-action is ever consulted --
# unless news itself is PENDING, in which case price-action is still
# evaluated next (out-of-order use, handled inside the PENDING branch).
if news.state != STATE_PENDING:
settled = _settle(STEP_NEWS, news)
if settled is not None:
status, entries = settled
missing.extend(entries)
return status, missing, extra_warnings
# news.state == CONFIRMED. Step 3: price-action.
if price.state == STATE_PENDING:
missing.append({"step": STEP_PRICE, "state": STATE_PENDING, "reason": "pending_step"})
return "WATCHING_PRICE", missing, extra_warnings
settled = _settle(STEP_PRICE, price)
if settled is not None:
status, entries = settled
missing.extend(entries)
return status, missing, extra_warnings
# price.state == CONFIRMED: all three confirmed.
return "READY_FOR_PLAN", missing, extra_warnings
# news.state == PENDING: out-of-order use. Price-action is still fully
# evaluated -- INVALID/NOT_CONFIRMED/INSUFFICIENT resolve normally; a
# CONFIRMED price-action caps the status at CROWDED with a warning
# (news was never actually confirmed, so this can't advance further).
missing.append({"step": STEP_NEWS, "state": STATE_PENDING, "reason": "pending_step"})
settled = _settle(STEP_PRICE, price)
if settled is not None:
status, entries = settled
missing.extend(entries)
return status, missing, extra_warnings
if price.state == STATE_CONFIRMED:
extra_warnings.append("out_of_order_price_action")
return "CROWDED", missing, extra_warnings
# price.state == PENDING too: crowding only.
missing.append({"step": STEP_PRICE, "state": STATE_PENDING, "reason": "pending_step"})
return "CROWDED", missing, extra_warnings
def _weakest_confidence(a: str | None, b: str | None) -> str | None:
values = [v for v in (a, b) if v is not None]
if not values:
return None
return min(values, key=lambda v: CONFIDENCE_RANK.get(v, 0))
def build_gate_result(
*,
symbol: str,
crowding: NormalizedInput,
news: NormalizedInput,
price: NormalizedInput,
max_detector_age_days: int,
max_report_age_days: int,
as_of: str,
) -> dict[str, Any]:
"""Compose the full output contract (plan §2) from three normalized
inputs. Pure: no I/O, deterministic given its inputs."""
setup_status, missing_confirmations, extra_warnings = decide_setup_status(crowding, news, price)
direction = crowding.direction if crowding.state == STATE_CONFIRMED else None
gate_confidence: str | None = None
entry_trigger: str | None = None
invalidation_level: float | None = None
if setup_status == "READY_FOR_PLAN":
gate_confidence = _weakest_confidence(news.confidence, price.confidence)
entry_trigger = price.entry_trigger
invalidation_level = price.stop_reference
# Defensive invariant (PR #249 user-review round 2, P1-A/P1-B):
# normalize_price_action's CONFIRMED path is the actual enforcer
# -- it never returns CONFIRMED without a validated entry_trigger
# and a finite positive stop_reference -- but this assert exists
# as a second, independent guard against a future regression that
# could let READY_FOR_PLAN slip out without both being usable.
assert isinstance(entry_trigger, str) and entry_trigger, (
"invariant violated: READY_FOR_PLAN requires a non-empty entry_trigger"
)
assert (
isinstance(invalidation_level, (int, float))
and not isinstance(invalidation_level, bool)
and math.isfinite(invalidation_level)
and invalidation_level > 0
), "invariant violated: READY_FOR_PLAN requires a finite, positive invalidation_level"
warnings = list(extra_warnings)
warnings.extend(crowding.warnings)
warnings.extend(news.warnings)
warnings.extend(price.warnings)
if setup_status == "READY_FOR_PLAN":
if price.confidence == "MEDIUM":
warnings.append("price_action_confidence_medium")
if news.confidence == "MEDIUM":
warnings.append("news_confidence_medium")
return {
"schema_version": SCHEMA_VERSION,
"symbol": symbol,
"setup_status": setup_status,
"direction": direction,
"gate_confidence": gate_confidence,
"entry_trigger": entry_trigger,
"invalidation_level": invalidation_level,
"missing_confirmations": missing_confirmations,
"warnings": warnings,
"inputs": {
STEP_CROWDING: crowding.to_audit_dict(),
STEP_NEWS: news.to_audit_dict(),
STEP_PRICE: price.to_audit_dict(),
},
"run_context": {
"symbol": symbol,
"as_of": as_of,
"max_detector_age_days": max_detector_age_days,
"max_report_age_days": max_report_age_days,
"schema_version": SCHEMA_VERSION,
"skill": SKILL_NAME,
},
}
scripts/run_contrarian_setup_gate.py
#!/usr/bin/env python3
"""
Contrarian Setup Gate -- CLI
The center of Jason Shapiro's contrarian pipeline (steps 1-3 already run):
reads the three upstream report JSONs -- cot-contrarian-detector (crowding,
required), news-reaction-failure-analyzer (news, optional), and
technical-analyst's contrarian-confirmation mode (price action, optional) --
and synthesizes them into one actionable `setup_status` via the fail-closed
state machine in gate_logic.py.
PURE and OFFLINE: no network, no API keys, no environment reads beyond the
three local report files this CLI is pointed at. The CLI's own job is
narrow: load the three JSON files with hardening -- unreadable / parse_error
/ non_finite are caught here in `load_json_file` (non_finite: a whole-file
iterative scan rejecting any non-finite float anywhere, even in a field the
gate never reads -- PR #249 user-review round 3; this closes off the only
route by which a raw non-finite value could otherwise reach the JSON
writer's `allow_nan=False` via an audit-echo field and crash with exit 1
on an entirely different code path than the one that determined the input
was bad); malformed / stale / and the remaining field-level classes are
detected inside gate_logic.normalize_*, which degrades a wrong-shaped or
otherwise-invalid report to INVALID rather than crashing -- then call into
gate_logic.py to normalize and synthesize, then write the JSON/Markdown
report. It never crashes on a bad input file: every run exits 0 and writes
a report, with the specific reason named in the failed input's
`inputs.<step>.state` / `missing_confirmations` (identical fail-closed
contract to every upstream skill in this pipeline).
Output:
- JSON: contrarian_setup_gate_<symbol>_<as-of>.json
- Markdown: contrarian_setup_gate_<symbol>_<as-of>.md
"""
from __future__ import annotations
import argparse
import json
import math
import sys
from datetime import datetime
from pathlib import Path
from typing import Any
import gate_logic
SKILL_NAME = "contrarian-setup-gate"
def _contains_non_finite(value: Any) -> bool:
"""Iteratively check whether a JSON-parsed structure contains any
non-finite float (`inf`, `-inf`, `nan`) ANYWHERE -- at any depth, in
any dict value or list item, regardless of whether that field is one
the gate actually reads. Catches both a syntactically valid JSON
number that overflows to `inf` on parse (e.g. `1e309` -- `math.isinf`
catches this even though no `parse_constant` hook ever sees it, since
the overflow happens inside `float()` itself, after the token has
already been recognized as an ordinary number) and the bare
`Infinity`/`-Infinity`/`NaN` literals `json.loads` accepts by default
as a non-standard extension (PR #249 user-review round 3).
Uses an explicit stack, NOT recursion: an ordinary, legitimate JSON
document can be nested hundreds of levels deep (all finite, in a
field the gate never even reads) and is perfectly valid input -- a
recursive walker raises `RecursionError` well before that (confirmed
empirically: a plain recursive version of this function fails
starting around depth ~500, since each JSON nesting level costs more
than one Python call frame once the `any(... for ...)` generator
expressions are counted), which used to crash the CLI with exit 1 on
a file that should have produced a completely NORMAL result (PR #249
user-review round 4). The iterative form has no depth limit -- it is
bounded only by available memory, not the call stack -- and is O(n)
over the total number of scalar/container nodes in the structure."""
stack: list[Any] = [value]
while stack:
current = stack.pop()
if isinstance(current, float):
if not math.isfinite(current):
return True
elif isinstance(current, dict):
stack.extend(current.values())
elif isinstance(current, list):
stack.extend(current)
return False
def load_json_file(path: str) -> tuple[Any | None, str | None]:
"""Read and parse a JSON file. Returns (data, None) on success, or
(None, reason) on failure, where `reason` is `"unreadable"` (missing,
permission-denied, a directory, or not valid UTF-8), `"parse_error"`
(the file opened and decoded fine but isn't valid JSON), or
`"non_finite"` (the file parsed as valid JSON, but contains a
non-finite float somewhere -- see `_contains_non_finite`).
Modeled on technical-analyst's post-#247 hardened loader
(skills/technical-analyst/scripts/check_weekly_price_action.py
load_json_file, the current merged version -- NOT
news-reaction-failure-analyzer's older 2-tuple OSError-only loader,
which still crashes on non-UTF-8 input; see issue #248). Catches
`(OSError, UnicodeError)`, not `OSError` alone: a readable file that
isn't valid UTF-8 raises `UnicodeDecodeError`, a `UnicodeError` /
`ValueError` subclass, NOT an `OSError` -- it would otherwise escape
this function entirely and crash with a traceback. `MemoryError` and
other unrelated exceptions are deliberately left uncaught -- those
aren't "this input is bad" conditions.
The `non_finite` check runs BEFORE this data is handed to
gate_logic.normalize_* at all -- it is deliberately whole-file, not
field-scoped: gate_logic's normalize_* functions preserve some
raw-but-invalid values (e.g. an unknown `classification`) verbatim in
their `NormalizedInput` for audit transparency, and the CLI's JSON
writer uses `allow_nan=False` as a second, independent defense layer
-- a non-finite float that reached that echo path would otherwise
raise `ValueError` at write time and break the exit-0-always-writes-
a-report contract on an entirely different path than the one that
determined the input was bad (PR #249 user-review round 3: this
scenario was reproduced live -- `classification: 1e309` normalized
correctly to INVALID, but then crashed the CLI with exit 1 when the
raw `inf` value was echoed into the audit block and hit `allow_nan=
False`). Scanning the whole parsed structure up front, before
gate_logic ever sees it, closes that off structurally rather than
requiring every future echo site to remember to guard itself.
Wrong-shape JSON (valid JSON, but not a dict / missing required keys /
wrong types) is NOT this function's concern -- that's detected inside
gate_logic.normalize_*, which degrades to an INVALID state with a
named `<input>_malformed` reason rather than raising here.
"""
try:
text = Path(path).read_text(encoding="utf-8")
except (OSError, UnicodeError):
return None, "unreadable"
try:
data = json.loads(text)
except json.JSONDecodeError:
return None, "parse_error"
except RecursionError:
# json.loads itself -- not just our own scanner -- can raise this
# on extreme nesting. Confirmed empirically in this environment:
# the C accelerator handles depth 100,000 fine but raises
# RecursionError around depth 200,000 ("Stack overflow ... while
# decoding a JSON array"). RecursionError is NOT a ValueError /
# JSONDecodeError subclass, so it must be caught explicitly, or
# it escapes this function entirely and crashes the CLI (PR #249
# user-review round 4). Routed to the same structured
# `parse_error` class as any other unparsable input -- exit 0,
# a report is still written, named `<input>_parse_error`.
return None, "parse_error"
if _contains_non_finite(data):
return None, "non_finite"
return data, None
def _as_of_type(value: str) -> str:
"""argparse type for --as-of: must be YYYY-MM-DD. This validates an
OPERATOR-supplied CLI argument (not one of the three untrusted report
files), so a bad value is a genuine usage error, not a degraded-input
case -- argparse fails the run before any file is even opened."""
try:
datetime.strptime(value, "%Y-%m-%d")
except ValueError as exc:
raise argparse.ArgumentTypeError(f"--as-of must be YYYY-MM-DD, got {value!r}") from exc
return value
def parse_arguments() -> argparse.Namespace:
parser = argparse.ArgumentParser(
description="Synthesize the three Shapiro contrarian-pipeline verdicts "
"(crowding / news-failure / price-action) into one actionable setup_status",
formatter_class=argparse.RawDescriptionHelpFormatter,
)
parser.add_argument("--symbol", required=True, help="The market under evaluation, e.g. B6")
parser.add_argument(
"--detector-json", required=True, help="cot-contrarian-detector JSON report path (step 1)"
)
parser.add_argument(
"--news-json", help="news-reaction-failure-analyzer JSON report path (step 2, optional)"
)
parser.add_argument(
"--price-action-json",
help="technical-analyst contrarian-confirmation JSON report path (step 3, optional)",
)
parser.add_argument(
"--as-of", required=True, type=_as_of_type, help="Reference date for staleness (YYYY-MM-DD)"
)
parser.add_argument("--max-detector-age-days", type=int, default=10)
parser.add_argument("--max-report-age-days", type=int, default=7)
parser.add_argument("--output-dir", default="reports/")
parser.add_argument("--format", choices=["json", "md", "both"], default="both")
return parser.parse_args()
def generate_json_report(result: dict[str, Any], output_path: Path) -> None:
# allow_nan=False is defense-in-depth (PR #249 user-review round 2,
# P1-B): gate_logic's own validation should already reject any
# non-finite number before it reaches this payload, but if one ever
# did slip through, json.dumps would otherwise silently emit the
# non-standard `Infinity`/`-Infinity`/`NaN` literals instead of
# raising -- this makes that failure loud instead of producing an
# invalid JSON file.
output_path.write_text(
json.dumps(result, indent=2, sort_keys=False, allow_nan=False) + "\n", encoding="utf-8"
)
def _fmt(value: Any) -> str:
return "n/a" if value is None else str(value)
def generate_markdown_report(result: dict[str, Any], output_path: Path) -> None:
inputs = result.get("inputs", {})
missing = result.get("missing_confirmations", [])
warnings = result.get("warnings", [])
run_context = result.get("run_context", {})
lines = [
"# Contrarian Setup Gate",
"",
f"Symbol: {result.get('symbol')}",
f"As-of date: {run_context.get('as_of')}",
"",
"## Setup Status",
"",
f"**{result.get('setup_status')}**",
f"Direction: `{_fmt(result.get('direction'))}`",
f"Gate confidence: {_fmt(result.get('gate_confidence'))}",
f"Entry trigger: {_fmt(result.get('entry_trigger'))}",
f"Invalidation level: {_fmt(result.get('invalidation_level'))}",
"",
"## Pipeline Inputs",
"",
"| Step | State | Verdict/Classification | Confidence | Reason | Age (days) | Report |",
"|---|---|---|---|---|---|---|",
]
for step_key, label in (
("crowding", "1. Crowding (cot-contrarian-detector)"),
("news_failure", "2. News Failure (news-reaction-failure-analyzer)"),
("price_action", "3. Price Action (technical-analyst)"),
):
block = inputs.get(step_key, {})
verdict_or_classification = block.get("classification") or block.get("verdict")
reason = block.get("verdict_reason") or ""
lines.append(
f"| {label} | {block.get('state')} | {_fmt(verdict_or_classification)} | "
f"{_fmt(block.get('confidence'))} | {reason} | {_fmt(block.get('age_days'))} | "
f"{_fmt(block.get('report_path'))} |"
)
lines += ["", "## Missing / Blocking Confirmations", ""]
if missing:
lines += ["| Step | State | Reason |", "|---|---|---|"]
for item in missing:
lines.append(f"| {item.get('step')} | {item.get('state')} | {item.get('reason')} |")
else:
lines.append("None.")
lines += ["", "## Warnings", ""]
if warnings:
lines.extend(f"- {warning}" for warning in warnings)
else:
lines.append("None.")
lines += [
"",
"## Guardrails",
"",
"This gate never places or recommends an order. READY_FOR_PLAN is the "
"furthest this skill advances -- position sizing and order entry are "
"downstream decisions. INSUFFICIENT_EVIDENCE and REJECTED never advance "
"regardless of any other input. Not investment advice.",
"",
"## Methodology",
"",
"Synthesizes Jason Shapiro's 3-step contrarian process (COT crowding -> "
"news-reaction failure -> weekly price-action confirmation) via an "
"explicit precedence state machine. See "
"`references/gate-decision-table.md` for the full decision table.",
"",
]
output_path.write_text("\n".join(lines), encoding="utf-8")
def main() -> int:
args = parse_arguments()
print("=" * 72)
print("Contrarian Setup Gate")
print("=" * 72)
symbol = args.symbol.strip().upper()
output_dir = Path(args.output_dir)
output_dir.mkdir(parents=True, exist_ok=True)
json_path = output_dir / f"contrarian_setup_gate_{symbol}_{args.as_of}.json"
md_path = output_dir / f"contrarian_setup_gate_{symbol}_{args.as_of}.md"
detector_data, detector_load_error = load_json_file(args.detector_json)
crowding = gate_logic.normalize_crowding(
detector_data,
detector_load_error,
symbol=symbol,
as_of=args.as_of,
max_age_days=args.max_detector_age_days,
report_path=args.detector_json,
)
if args.news_json:
news_data, news_load_error = load_json_file(args.news_json)
news = gate_logic.normalize_news(
news_data,
news_load_error,
symbol=symbol,
as_of=args.as_of,
max_age_days=args.max_report_age_days,
detector=crowding,
report_path=args.news_json,
)
else:
news = gate_logic.pending_input(gate_logic.STEP_NEWS)
if args.price_action_json:
price_data, price_load_error = load_json_file(args.price_action_json)
price = gate_logic.normalize_price_action(
price_data,
price_load_error,
symbol=symbol,
as_of=args.as_of,
max_age_days=args.max_report_age_days,
detector=crowding,
report_path=args.price_action_json,
)
else:
price = gate_logic.pending_input(gate_logic.STEP_PRICE)
result = gate_logic.build_gate_result(
symbol=symbol,
crowding=crowding,
news=news,
price=price,
max_detector_age_days=args.max_detector_age_days,
max_report_age_days=args.max_report_age_days,
as_of=args.as_of,
)
if args.format in ("json", "both"):
generate_json_report(result, json_path)
if args.format in ("md", "both"):
generate_markdown_report(result, md_path)
print(f"Setup status: {result['setup_status']} (direction: {result['direction']})")
if result["missing_confirmations"]:
for item in result["missing_confirmations"]:
print(f" - {item['step']}: {item['state']} ({item['reason']})")
if args.format in ("json", "both"):
print(f" JSON Report: {json_path}")
if args.format in ("md", "both"):
print(f" Markdown Report: {md_path}")
# Every valid decision (including INSUFFICIENT_EVIDENCE and REJECTED)
# exits 0 -- those are correct, fail-closed outcomes, not errors.
return 0
if __name__ == "__main__":
try:
sys.exit(main())
except Exception as exc: # noqa: BLE001 - CLI should return a clear message, not a traceback.
print(f"Error: {exc}", file=sys.stderr)
sys.exit(1)
scripts/tests/conftest.py
"""Pytest configuration for contrarian-setup-gate tests."""
import sys
from pathlib import Path
scripts_dir = Path(__file__).resolve().parents[1]
if str(scripts_dir) not in sys.path:
sys.path.insert(0, str(scripts_dir))
scripts/tests/test_gate_logic.py
"""Tests for gate_logic.py -- the pure contrarian-setup-gate synthesis core.
Section 1 is the exhaustive state-machine matrix (plan Issue #241 §6's DoD:
"every verdict combination"), parametrized over all 125 reachable
combinations of normalized states, checked against an independent oracle
that encodes the precedence rules directly from the spec (not by calling
into decide_setup_status -- a matrix test that only re-derives the
implementation proves nothing). Section 2 pins the named precedence
combinations from the plan explicitly, for documentation and regression
value. Later sections cover normalization from real-schema fixtures,
consistency checks, staleness, warnings, and gate_confidence/entry_trigger.
"""
from __future__ import annotations
import sys
from pathlib import Path
import pytest
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
import gate_logic as gl # noqa: E402
SYMBOL = "B6"
AS_OF = "2026-07-15"
# --- Section 1: exhaustive state-machine matrix -----------------------------
C_LABELS = ("CONF_L", "CONF_S", "NOT_CONF", "INSUFF", "INVALID")
NP_LABELS = ("CONF", "NOT_CONF", "INSUFF", "PENDING", "INVALID")
def _crowding(label: str) -> gl.NormalizedInput:
if label == "CONF_L":
return gl.NormalizedInput(
kind=gl.STEP_CROWDING,
state=gl.STATE_CONFIRMED,
classification="CROWDED_LONG",
direction="SHORT",
)
if label == "CONF_S":
return gl.NormalizedInput(
kind=gl.STEP_CROWDING,
state=gl.STATE_CONFIRMED,
classification="CROWDED_SHORT",
direction="LONG",
)
if label == "NOT_CONF":
return gl.NormalizedInput(
kind=gl.STEP_CROWDING, state=gl.STATE_NOT_CONFIRMED, reason="detector_not_crowded"
)
if label == "INSUFF":
return gl.NormalizedInput(
kind=gl.STEP_CROWDING, state=gl.STATE_INSUFFICIENT, reason="detector_missing_symbol"
)
if label == "INVALID":
return gl.NormalizedInput(
kind=gl.STEP_CROWDING, state=gl.STATE_INVALID, reason="detector_json_stale"
)
raise AssertionError(label)
def _downstream(kind: str, label: str, prefix: str) -> gl.NormalizedInput:
if label == "CONF":
if kind == gl.STEP_PRICE:
# A CONFIRMED price-action step must carry a validated
# entry_trigger and stop_reference -- build_gate_result's
# READY_FOR_PLAN invariant assert enforces this even for this
# status-transition-only matrix (PR #249 user-review round 2).
return gl.NormalizedInput(
kind=kind,
state=gl.STATE_CONFIRMED,
confidence="HIGH",
entry_trigger="price-action confirmation: key_reversal at week_of=2026-07-06",
stop_reference=1.372,
)
return gl.NormalizedInput(kind=kind, state=gl.STATE_CONFIRMED, confidence="HIGH")
if label == "NOT_CONF":
return gl.NormalizedInput(
kind=kind, state=gl.STATE_NOT_CONFIRMED, reason=f"{prefix}_not_confirmed"
)
if label == "INSUFF":
return gl.NormalizedInput(
kind=kind, state=gl.STATE_INSUFFICIENT, reason=f"{prefix}_insufficient"
)
if label == "PENDING":
return gl.pending_input(kind)
if label == "INVALID":
return gl.NormalizedInput(kind=kind, state=gl.STATE_INVALID, reason=f"{prefix}_unreadable")
raise AssertionError(label)
def _news(label: str) -> gl.NormalizedInput:
return _downstream(gl.STEP_NEWS, label, "news")
def _price(label: str) -> gl.NormalizedInput:
return _downstream(gl.STEP_PRICE, label, "price_action")
def _oracle_status(c: str, n: str, p: str) -> str:
"""Independent re-statement of plan §3.3's v4 SEQUENTIAL precedence
(PR #249 P1-3), deliberately NOT derived from gate_logic's own code.
Each step fully settles before the next is even looked at -- news
settles before price-action is inspected at all, so a later step's
problem can never soften an earlier step's definitive verdict."""
if c in ("INVALID", "INSUFF"):
return "INSUFFICIENT_EVIDENCE"
if c == "NOT_CONF":
return "REJECTED"
# c in (CONF_L, CONF_S) from here. Step 2: news settles first, on its
# own, without even looking at p yet.
if n == "INVALID":
return "INSUFFICIENT_EVIDENCE"
if n == "NOT_CONF":
return "REJECTED"
if n == "INSUFF":
return "INSUFFICIENT_EVIDENCE"
if n == "PENDING":
# Step 3, out-of-order: p is still fully evaluated on its own.
if p == "INVALID":
return "INSUFFICIENT_EVIDENCE"
if p == "NOT_CONF":
return "REJECTED"
if p == "INSUFF":
return "INSUFFICIENT_EVIDENCE"
if p == "CONF":
return "CROWDED" # out-of-order cap + warning
return "CROWDED" # p == PENDING too: crowding only
# n == CONF. Step 3: price-action, evaluated on its own.
if p == "INVALID":
return "INSUFFICIENT_EVIDENCE"
if p == "NOT_CONF":
return "REJECTED"
if p == "INSUFF":
return "INSUFFICIENT_EVIDENCE"
if p == "PENDING":
return "WATCHING_PRICE"
return "READY_FOR_PLAN" # p == CONF: all three confirmed
@pytest.mark.parametrize("c_label", C_LABELS)
@pytest.mark.parametrize("n_label", NP_LABELS)
@pytest.mark.parametrize("p_label", NP_LABELS)
def test_state_machine_matrix(c_label: str, n_label: str, p_label: str) -> None:
crowding = _crowding(c_label)
news = _news(n_label)
price = _price(p_label)
expected_status = _oracle_status(c_label, n_label, p_label)
status, _missing, _warnings = gl.decide_setup_status(crowding, news, price)
assert status == expected_status, (c_label, n_label, p_label)
# direction: populated iff crowding is CONFIRMED, regardless of the
# overall status (plan §2: "null unless crowding usable & confirmed").
result = gl.build_gate_result(
symbol=SYMBOL,
crowding=crowding,
news=news,
price=price,
max_detector_age_days=10,
max_report_age_days=7,
as_of=AS_OF,
)
if c_label == "CONF_L":
assert result["direction"] == "SHORT"
elif c_label == "CONF_S":
assert result["direction"] == "LONG"
else:
assert result["direction"] is None
# gate_confidence/entry_trigger/invalidation_level: only at READY_FOR_PLAN.
if expected_status != "READY_FOR_PLAN":
assert result["gate_confidence"] is None
assert result["entry_trigger"] is None
assert result["invalidation_level"] is None
# --- Section 2: named precedence pins (plan §3.3, §6) -----------------------
def test_pin_not_confirmed_beats_downstream_unreadable_invalid() -> None:
"""C=NOT_CONFIRMED + N=INVALID(unreadable) -> REJECTED (P1-1): crowding's
own conclusion is never softened by a corrupted downstream file."""
crowding = _crowding("NOT_CONF")
news = gl.NormalizedInput(kind=gl.STEP_NEWS, state=gl.STATE_INVALID, reason="news_unreadable")
price = gl.pending_input(gl.STEP_PRICE)
status, missing, _warnings = gl.decide_setup_status(crowding, news, price)
assert status == "REJECTED"
assert missing == [
{"step": "crowding", "state": "NOT_CONFIRMED", "reason": "detector_not_crowded"}
]
def test_pin_not_confirmed_beats_downstream_symbol_mismatch_invalid() -> None:
"""C=NOT_CONFIRMED + N=INVALID(symbol_mismatch) -> REJECTED: per-input
INVALID from a consistency failure is never a global override that
could soften crowding's own conclusion (round-2 P1, v3)."""
crowding = _crowding("NOT_CONF")
news = gl.NormalizedInput(
kind=gl.STEP_NEWS, state=gl.STATE_INVALID, reason="news_symbol_mismatch"
)
price = gl.pending_input(gl.STEP_PRICE)
status, _missing, _warnings = gl.decide_setup_status(crowding, news, price)
assert status == "REJECTED"
def test_pin_not_confirmed_beats_price_not_confirmed() -> None:
"""C=NOT_CONFIRMED + P=NOT_CONFIRMED -> REJECTED (crowding still the
exclusive rejector; missing_confirmations names crowding, not price)."""
crowding = _crowding("NOT_CONF")
news = gl.pending_input(gl.STEP_NEWS)
price = gl.NormalizedInput(
kind=gl.STEP_PRICE, state=gl.STATE_NOT_CONFIRMED, reason="no_reversal_evidence"
)
status, missing, _warnings = gl.decide_setup_status(crowding, news, price)
assert status == "REJECTED"
assert missing == [
{"step": "crowding", "state": "NOT_CONFIRMED", "reason": "detector_not_crowded"}
]
def test_pin_crowding_invalid_beats_downstream_not_confirmed() -> None:
"""C=INVALID + N=NOT_CONFIRMED -> INSUFFICIENT_EVIDENCE (rule 1 first)."""
crowding = _crowding("INVALID")
news = gl.NormalizedInput(kind=gl.STEP_NEWS, state=gl.STATE_NOT_CONFIRMED, reason="x")
price = gl.pending_input(gl.STEP_PRICE)
status, missing, _warnings = gl.decide_setup_status(crowding, news, price)
assert status == "INSUFFICIENT_EVIDENCE"
assert missing == [{"step": "crowding", "state": "INVALID", "reason": "detector_json_stale"}]
def test_pin_crowding_insufficient_beats_downstream_not_confirmed() -> None:
"""C=INSUFFICIENT + N=NOT_CONFIRMED -> INSUFFICIENT_EVIDENCE."""
crowding = _crowding("INSUFF")
news = gl.NormalizedInput(kind=gl.STEP_NEWS, state=gl.STATE_NOT_CONFIRMED, reason="x")
price = gl.pending_input(gl.STEP_PRICE)
status, _missing, _warnings = gl.decide_setup_status(crowding, news, price)
assert status == "INSUFFICIENT_EVIDENCE"
def test_pin_news_invalid_settles_before_price_is_ever_consulted() -> None:
"""C=CONFIRMED + N=INVALID + P=NOT_CONFIRMED -> INSUFFICIENT_EVIDENCE:
news (step 2) settles as INVALID and returns immediately -- price
(step 3) is never even inspected, so its NOT_CONFIRMED plays no part
(v4 sequential model, PR #249 P1-3)."""
crowding = _crowding("CONF_S")
news = gl.NormalizedInput(kind=gl.STEP_NEWS, state=gl.STATE_INVALID, reason="news_parse_error")
price = gl.NormalizedInput(kind=gl.STEP_PRICE, state=gl.STATE_NOT_CONFIRMED, reason="x")
status, missing, _warnings = gl.decide_setup_status(crowding, news, price)
assert status == "INSUFFICIENT_EVIDENCE"
# Only the deciding (INVALID) step is named -- price was never reached.
assert missing == [{"step": "news_failure", "state": "INVALID", "reason": "news_parse_error"}]
def test_pin_news_not_confirmed_settles_before_price_is_ever_consulted() -> None:
"""C=CONFIRMED + N=NOT_CONFIRMED + P=INSUFFICIENT -> REJECTED: news
settles as NOT_CONFIRMED and returns immediately -- price's own
INSUFFICIENT state is never reached (v4 sequential model)."""
crowding = _crowding("CONF_L")
news = gl.NormalizedInput(kind=gl.STEP_NEWS, state=gl.STATE_NOT_CONFIRMED, reason="x")
price = gl.NormalizedInput(kind=gl.STEP_PRICE, state=gl.STATE_INSUFFICIENT, reason="y")
status, missing, _warnings = gl.decide_setup_status(crowding, news, price)
assert status == "REJECTED"
assert missing == [{"step": "news_failure", "state": "NOT_CONFIRMED", "reason": "x"}]
def test_pin_news_not_confirmed_beats_price_invalid_p1_3_repro() -> None:
"""THE USER'S PR #249 P1-3 REPRO: C=CONFIRMED + N=NOT_CONFIRMED +
P=INVALID(unreadable/binary) -> REJECTED, not INSUFFICIENT_EVIDENCE.
Before the v4 fix, the v3 aggregate rule ("any provided INVALID
across N/P" checked before "any NOT_CONFIRMED across N/P") let a
LATER step's file corruption soften an EARLIER step's definitive
rejection. News settling first structurally prevents that: price is
never even inspected once news has already rejected."""
crowding = _crowding("CONF_S")
news = gl.NormalizedInput(kind=gl.STEP_NEWS, state=gl.STATE_NOT_CONFIRMED, reason="no_reversal")
price = gl.NormalizedInput(
kind=gl.STEP_PRICE, state=gl.STATE_INVALID, reason="price_action_unreadable"
)
status, missing, _warnings = gl.decide_setup_status(crowding, news, price)
assert status == "REJECTED"
assert missing == [{"step": "news_failure", "state": "NOT_CONFIRMED", "reason": "no_reversal"}]
def test_pin_news_insufficient_beats_price_not_confirmed_symmetric_case() -> None:
"""Symmetric to the P1-3 repro: C=CONFIRMED + N=INSUFFICIENT +
P=NOT_CONFIRMED -> INSUFFICIENT_EVIDENCE, not REJECTED. News settles
first as INSUFFICIENT (a run that tried and could not judge never
advances); price's own NOT_CONFIRMED is never reached to reject."""
crowding = _crowding("CONF_L")
news = gl.NormalizedInput(
kind=gl.STEP_NEWS, state=gl.STATE_INSUFFICIENT, reason="no_usable_events"
)
price = gl.NormalizedInput(
kind=gl.STEP_PRICE, state=gl.STATE_NOT_CONFIRMED, reason="no_reversal"
)
status, missing, _warnings = gl.decide_setup_status(crowding, news, price)
assert status == "INSUFFICIENT_EVIDENCE"
assert missing == [
{"step": "news_failure", "state": "INSUFFICIENT", "reason": "no_usable_events"}
]
def test_pin_out_of_order_price_without_news_caps_at_crowded() -> None:
crowding = _crowding("CONF_S")
news = gl.pending_input(gl.STEP_NEWS)
price = gl.NormalizedInput(kind=gl.STEP_PRICE, state=gl.STATE_CONFIRMED, confidence="HIGH")
status, missing, warnings = gl.decide_setup_status(crowding, news, price)
assert status == "CROWDED"
assert "out_of_order_price_action" in warnings
assert {"step": "news_failure", "state": "PENDING", "reason": "pending_step"} in missing
def test_pin_out_of_order_price_not_confirmed_without_news_still_rejects() -> None:
"""A NOT_CONFIRMED price verdict still REJECTs even when news is
PENDING (out-of-order use) -- rule 3 runs before rule 5's capping."""
crowding = _crowding("CONF_S")
news = gl.pending_input(gl.STEP_NEWS)
price = gl.NormalizedInput(
kind=gl.STEP_PRICE, state=gl.STATE_NOT_CONFIRMED, reason="no_reversal_evidence"
)
status, _missing, warnings = gl.decide_setup_status(crowding, news, price)
assert status == "REJECTED"
assert "out_of_order_price_action" not in warnings
# --- Section 3: normalize_crowding from real-schema fixtures ---------------
def detector_fixture(
*,
symbol: str = SYMBOL,
classification: str = "CROWDED_SHORT",
data_date: str = "2026-07-07",
run_context_data_date: str | None = "2026-07-07",
schema_version: str = "1.0",
skipped: list | None = None,
extra_market_fields: dict | None = None,
) -> dict:
"""Mirrors the real cot-contrarian-detector output shape (verified
against reports/cot_crowding_2026-07-12.json, regenerated live)."""
market_row = {
"symbol": symbol,
"status": "ok",
"name": f"{symbol} Futures",
"data_date": data_date,
"classification": classification,
"cot_index_3y": 7.2,
"net_position": -87903,
}
if extra_market_fields:
market_row.update(extra_market_fields)
return {
"schema_version": schema_version,
"skill": "cot-contrarian-detector",
"run_context": {
"schema_version": schema_version,
"skill": "cot-contrarian-detector",
"as_of": "2026-07-12",
"data_date": run_context_data_date,
},
"markets": [market_row],
"skipped": skipped or [],
}
def test_normalize_crowding_confirmed_crowded_short() -> None:
data = detector_fixture(classification="CROWDED_SHORT")
result = gl.normalize_crowding(data, None, symbol=SYMBOL, as_of=AS_OF, max_age_days=10)
assert result.state == gl.STATE_CONFIRMED
assert result.classification == "CROWDED_SHORT"
assert result.direction == "LONG"
assert result.age_days == 8
def test_normalize_crowding_confirmed_crowded_long() -> None:
data = detector_fixture(symbol="BT", classification="CROWDED_LONG")
result = gl.normalize_crowding(data, None, symbol="BT", as_of=AS_OF, max_age_days=10)
assert result.state == gl.STATE_CONFIRMED
assert result.direction == "SHORT"
def test_normalize_crowding_neutral_is_not_confirmed() -> None:
data = detector_fixture(classification="NEUTRAL")
result = gl.normalize_crowding(data, None, symbol=SYMBOL, as_of=AS_OF, max_age_days=10)
assert result.state == gl.STATE_NOT_CONFIRMED
assert result.reason == "detector_not_crowded"
def test_normalize_crowding_symbol_absent_is_insufficient() -> None:
data = detector_fixture(symbol="ZZ")
result = gl.normalize_crowding(data, None, symbol=SYMBOL, as_of=AS_OF, max_age_days=10)
assert result.state == gl.STATE_INSUFFICIENT
assert result.reason == "detector_missing_symbol"
def test_normalize_crowding_symbol_in_skipped_is_insufficient() -> None:
data = detector_fixture()
data["skipped"] = [{"symbol": SYMBOL, "reason": "no data returned by API"}]
result = gl.normalize_crowding(data, None, symbol=SYMBOL, as_of=AS_OF, max_age_days=10)
assert result.state == gl.STATE_INSUFFICIENT
assert result.reason == "detector_missing_symbol"
def test_normalize_crowding_skipped_entry_with_unhashable_symbol_never_crashes() -> None:
"""PR #249 P1-1: an untrusted skip-entry whose `symbol` is a JSON list
must not crash building the skipped-symbols set (adding an unhashable
value to a set raises TypeError). The bad entry is simply excluded
from the set; the market row (present and usable) still normalizes."""
data = detector_fixture(classification="CROWDED_SHORT")
data["skipped"] = [{"symbol": ["not", "hashable"], "reason": "malformed"}]
result = gl.normalize_crowding(data, None, symbol=SYMBOL, as_of=AS_OF, max_age_days=10)
assert result.state == gl.STATE_CONFIRMED
def test_normalize_crowding_duplicate_symbol_rows_first_match_wins() -> None:
"""Duplicate `symbol` rows in markets[] -> first match wins (v2 P2-7)."""
data = detector_fixture(classification="CROWDED_SHORT")
data["markets"].append(
{
"symbol": SYMBOL,
"status": "ok",
"data_date": "2026-07-07",
"classification": "CROWDED_LONG",
}
)
result = gl.normalize_crowding(data, None, symbol=SYMBOL, as_of=AS_OF, max_age_days=10)
assert result.classification == "CROWDED_SHORT"
@pytest.mark.parametrize(
"data_date_value,expected_reason",
[
(None, "detector_missing_data_date"),
("", "detector_missing_data_date"),
(20260707, "detector_invalid_data_date"),
("not-a-date", "detector_invalid_data_date"),
("2026-07-20", "detector_future_data_date"), # after AS_OF 2026-07-15
],
)
def test_normalize_crowding_data_date_guards(data_date_value, expected_reason) -> None:
data = detector_fixture(run_context_data_date=data_date_value)
result = gl.normalize_crowding(data, None, symbol=SYMBOL, as_of=AS_OF, max_age_days=10)
assert result.state == gl.STATE_INVALID
assert result.reason == expected_reason
def test_normalize_crowding_stale_beyond_max_age() -> None:
data = detector_fixture(run_context_data_date="2026-06-01")
result = gl.normalize_crowding(data, None, symbol=SYMBOL, as_of=AS_OF, max_age_days=10)
assert result.state == gl.STATE_INVALID
assert result.reason == "detector_json_stale"
def test_normalize_crowding_near_stale_warning() -> None:
# AS_OF 2026-07-15, data_date 2026-07-06 -> age 9, max 10 -> within 2.
data = detector_fixture(run_context_data_date="2026-07-06")
result = gl.normalize_crowding(data, None, symbol=SYMBOL, as_of=AS_OF, max_age_days=10)
assert result.state == gl.STATE_CONFIRMED
assert "detector_near_stale" in result.warnings
def test_normalize_crowding_data_date_divergence_warning() -> None:
data = detector_fixture(data_date="2026-07-01", run_context_data_date="2026-07-07")
result = gl.normalize_crowding(data, None, symbol=SYMBOL, as_of=AS_OF, max_age_days=10)
assert "detector_data_date_divergence" in result.warnings
def test_normalize_crowding_unknown_classification() -> None:
data = detector_fixture(classification="WEIRD")
result = gl.normalize_crowding(data, None, symbol=SYMBOL, as_of=AS_OF, max_age_days=10)
assert result.state == gl.STATE_INVALID
assert result.reason == "detector_unknown_classification"
@pytest.mark.parametrize("bad_classification", [["CROWDED_LONG"], {"a": 1}, 123, True])
def test_normalize_crowding_unhashable_or_wrong_type_classification_never_crashes(
bad_classification,
) -> None:
"""PR #249 P1-1: `classification not in FADE_DIRECTION` requires a
hashable operand -- an unhashable classification (list/dict) used to
raise TypeError instead of failing closed. Every non-string type must
degrade to detector_unknown_classification, never crash."""
data = detector_fixture(classification=bad_classification)
result = gl.normalize_crowding(data, None, symbol=SYMBOL, as_of=AS_OF, max_age_days=10)
assert result.state == gl.STATE_INVALID
assert result.reason == "detector_unknown_classification"
def test_normalize_crowding_schema_unsupported() -> None:
data = detector_fixture(schema_version="2.0")
result = gl.normalize_crowding(data, None, symbol=SYMBOL, as_of=AS_OF, max_age_days=10)
assert result.state == gl.STATE_INVALID
assert result.reason == "detector_schema_unsupported"
def test_normalize_crowding_load_error_unreadable() -> None:
result = gl.normalize_crowding(None, "unreadable", symbol=SYMBOL, as_of=AS_OF, max_age_days=10)
assert result.state == gl.STATE_INVALID
assert result.reason == "detector_unreadable"
def test_normalize_crowding_load_error_parse_error() -> None:
result = gl.normalize_crowding(None, "parse_error", symbol=SYMBOL, as_of=AS_OF, max_age_days=10)
assert result.state == gl.STATE_INVALID
assert result.reason == "detector_parse_error"
def test_normalize_crowding_load_error_non_finite() -> None:
"""PR #249 user-review round 3: the CLI's whole-file non-finite scan
reports this as a load_error, exactly like unreadable/parse_error --
gate_logic never even sees the raw parsed data in this case."""
result = gl.normalize_crowding(None, "non_finite", symbol=SYMBOL, as_of=AS_OF, max_age_days=10)
assert result.state == gl.STATE_INVALID
assert result.reason == "detector_non_finite"
@pytest.mark.parametrize("bad_shape", [["not", "a", "dict"], "a string", None, 42])
def test_normalize_crowding_wrong_top_level_shape(bad_shape) -> None:
result = gl.normalize_crowding(bad_shape, None, symbol=SYMBOL, as_of=AS_OF, max_age_days=10)
assert result.state == gl.STATE_INVALID
assert result.reason == "detector_malformed"
# --- Section 4: normalize_news / normalize_price_action ---------------------
CONFIRMED_CROWDING = gl.NormalizedInput(
kind=gl.STEP_CROWDING,
state=gl.STATE_CONFIRMED,
classification="CROWDED_SHORT",
direction="LONG",
)
UNUSABLE_CROWDING = gl.NormalizedInput(
kind=gl.STEP_CROWDING, state=gl.STATE_INVALID, reason="detector_json_stale"
)
def news_fixture(
*,
symbol: str = SYMBOL,
direction: str = "CROWDED_SHORT",
verdict: str = "NOT_CONFIRMED",
confidence: str = "HIGH",
verdict_reason: str = "no_significant_drift",
as_of: str = "2026-07-13",
schema_version: str = "1.0",
) -> dict:
"""Mirrors the real news-reaction-failure-analyzer output shape
(verified against analyze_news_reaction.py's output builder)."""
return {
"schema_version": schema_version,
"skill": "news-reaction-failure-analyzer",
"symbol": symbol,
"direction": direction,
"expected_direction": "BULLISH",
"actual_reaction": "NO_REACTION",
"verdict": verdict,
"confidence": confidence,
"relevant_events_used": 3,
"aggregate": {"mean_z3": 0.1, "drift_stat": 0.2, "responded_ratio": 0.33},
"evidence": [],
"clusters": [],
"dropped_events": [],
"verdict_reason": verdict_reason,
"run_context": {"as_of": as_of, "price_symbol": "GBPUSD"},
}
def test_normalize_news_confirmed() -> None:
data = news_fixture(verdict="CONFIRMED", verdict_reason=None, confidence="HIGH")
result = gl.normalize_news(
data, None, symbol=SYMBOL, as_of=AS_OF, max_age_days=7, detector=CONFIRMED_CROWDING
)
assert result.state == gl.STATE_CONFIRMED
assert result.confidence == "HIGH"
assert result.age_days == 2
def test_normalize_news_not_confirmed_carries_upstream_reason() -> None:
data = news_fixture(verdict="NOT_CONFIRMED", verdict_reason="no_significant_drift")
result = gl.normalize_news(
data, None, symbol=SYMBOL, as_of=AS_OF, max_age_days=7, detector=CONFIRMED_CROWDING
)
assert result.state == gl.STATE_NOT_CONFIRMED
assert result.reason == "no_significant_drift"
def test_normalize_news_insufficient_evidence() -> None:
data = news_fixture(verdict="INSUFFICIENT_EVIDENCE", verdict_reason="no_usable_events")
result = gl.normalize_news(
data, None, symbol=SYMBOL, as_of=AS_OF, max_age_days=7, detector=CONFIRMED_CROWDING
)
assert result.state == gl.STATE_INSUFFICIENT
assert result.reason == "no_usable_events"
def test_normalize_news_symbol_mismatch() -> None:
data = news_fixture(symbol="D6")
result = gl.normalize_news(
data, None, symbol=SYMBOL, as_of=AS_OF, max_age_days=7, detector=CONFIRMED_CROWDING
)
assert result.state == gl.STATE_INVALID
assert result.reason == "news_symbol_mismatch"
def test_normalize_news_direction_mismatch_only_checked_when_detector_usable() -> None:
data = news_fixture(direction="CROWDED_LONG") # detector says CROWDED_SHORT
result = gl.normalize_news(
data, None, symbol=SYMBOL, as_of=AS_OF, max_age_days=7, detector=CONFIRMED_CROWDING
)
assert result.state == gl.STATE_INVALID
assert result.reason == "news_direction_mismatch"
# Same mismatched data, but detector is unusable -> check is skipped.
result_skipped = gl.normalize_news(
data, None, symbol=SYMBOL, as_of=AS_OF, max_age_days=7, detector=UNUSABLE_CROWDING
)
assert result_skipped.reason != "news_direction_mismatch"
def test_normalize_news_missing_direction_key_is_malformed() -> None:
data = news_fixture()
del data["direction"]
result = gl.normalize_news(
data, None, symbol=SYMBOL, as_of=AS_OF, max_age_days=7, detector=CONFIRMED_CROWDING
)
assert result.state == gl.STATE_INVALID
assert result.reason == "news_malformed"
def test_normalize_news_null_direction_is_insufficient_not_mismatch() -> None:
"""A provided `direction: null` is the report's OWN fail-closed exit
(e.g. NRF's no_direction_provided early exit), not a mismatch against
the detector's classification -- None != CROWDED_SHORT would otherwise
always be truthy and misreport this as news_direction_mismatch (P3-1).
"""
data = news_fixture(
direction=None, verdict="INSUFFICIENT_EVIDENCE", verdict_reason="no_direction_provided"
)
result = gl.normalize_news(
data, None, symbol=SYMBOL, as_of=AS_OF, max_age_days=7, detector=CONFIRMED_CROWDING
)
assert result.state == gl.STATE_INSUFFICIENT
assert result.reason == "no_direction_provided"
def test_normalize_news_null_direction_without_verdict_reason_is_malformed() -> None:
data = news_fixture(direction=None, verdict_reason=None)
result = gl.normalize_news(
data, None, symbol=SYMBOL, as_of=AS_OF, max_age_days=7, detector=CONFIRMED_CROWDING
)
assert result.state == gl.STATE_INVALID
assert result.reason == "news_malformed"
def test_normalize_news_schema_unsupported() -> None:
data = news_fixture(schema_version="2.0")
result = gl.normalize_news(
data, None, symbol=SYMBOL, as_of=AS_OF, max_age_days=7, detector=CONFIRMED_CROWDING
)
assert result.state == gl.STATE_INVALID
assert result.reason == "news_schema_unsupported"
def test_normalize_news_unknown_verdict() -> None:
data = news_fixture(verdict="MAYBE")
result = gl.normalize_news(
data, None, symbol=SYMBOL, as_of=AS_OF, max_age_days=7, detector=CONFIRMED_CROWDING
)
assert result.state == gl.STATE_INVALID
assert result.reason == "news_unknown_verdict"
def test_normalize_news_unhashable_verdict_never_crashes() -> None:
"""THE USER'S PR #249 P1-1 REPRO: verdict: [] used to raise TypeError
(`verdict not in valid_verdicts` requires a hashable operand) instead
of failing closed with exit 0 + a report."""
data = news_fixture(verdict=[])
result = gl.normalize_news(
data, None, symbol=SYMBOL, as_of=AS_OF, max_age_days=7, detector=CONFIRMED_CROWDING
)
assert result.state == gl.STATE_INVALID
assert result.reason == "news_malformed"
@pytest.mark.parametrize("bad_verdict", [{"a": 1}, 123, True, None])
def test_normalize_news_wrong_type_verdict_is_malformed(bad_verdict) -> None:
data = news_fixture(verdict=bad_verdict)
result = gl.normalize_news(
data, None, symbol=SYMBOL, as_of=AS_OF, max_age_days=7, detector=CONFIRMED_CROWDING
)
assert result.state == gl.STATE_INVALID
assert result.reason == "news_malformed"
def test_normalize_news_unhashable_confidence_never_crashes() -> None:
"""THE USER'S PR #249 P1-1 REPRO: confidence: {} used to raise
TypeError deep inside gate_confidence's CONFIDENCE_RANK.get(...)."""
data = news_fixture(confidence={})
result = gl.normalize_news(
data, None, symbol=SYMBOL, as_of=AS_OF, max_age_days=7, detector=CONFIRMED_CROWDING
)
assert result.state == gl.STATE_INVALID
assert result.reason == "news_malformed"
def test_normalize_news_unknown_confidence_string_fails_closed() -> None:
"""PR #249 P1-2 REPRO: confidence: "BANANA" used to pass straight
through to gate_confidence unvalidated instead of failing closed."""
data = news_fixture(verdict="CONFIRMED", confidence="BANANA")
result = gl.normalize_news(
data, None, symbol=SYMBOL, as_of=AS_OF, max_age_days=7, detector=CONFIRMED_CROWDING
)
assert result.state == gl.STATE_INVALID
assert result.reason == "news_unknown_confidence"
def test_normalize_news_low_confidence_is_accepted_reserved_token() -> None:
"""LOW is a reserved token both upstreams document but never emit --
accepted here, not rejected as unknown (PR #249 P1-2)."""
data = news_fixture(verdict="CONFIRMED", confidence="LOW")
result = gl.normalize_news(
data, None, symbol=SYMBOL, as_of=AS_OF, max_age_days=7, detector=CONFIRMED_CROWDING
)
assert result.state == gl.STATE_CONFIRMED
assert result.confidence == "LOW"
def test_normalize_news_non_string_direction_is_malformed_not_mismatch() -> None:
"""A non-null, non-string direction (e.g. a stray integer) is
malformed input, not a legitimate mismatch to compare against the
detector's classification (PR #249 P1-1)."""
data = news_fixture(direction=123)
result = gl.normalize_news(
data, None, symbol=SYMBOL, as_of=AS_OF, max_age_days=7, detector=CONFIRMED_CROWDING
)
assert result.state == gl.STATE_INVALID
assert result.reason == "news_malformed"
def test_normalize_news_stale() -> None:
data = news_fixture(as_of="2026-07-01") # 14 days before AS_OF, max 7
result = gl.normalize_news(
data, None, symbol=SYMBOL, as_of=AS_OF, max_age_days=7, detector=CONFIRMED_CROWDING
)
assert result.state == gl.STATE_INVALID
assert result.reason == "news_json_stale"
def test_normalize_news_load_error_unreadable() -> None:
result = gl.normalize_news(
None, "unreadable", symbol=SYMBOL, as_of=AS_OF, max_age_days=7, detector=CONFIRMED_CROWDING
)
assert result.state == gl.STATE_INVALID
assert result.reason == "news_unreadable"
def test_normalize_news_load_error_non_finite() -> None:
result = gl.normalize_news(
None, "non_finite", symbol=SYMBOL, as_of=AS_OF, max_age_days=7, detector=CONFIRMED_CROWDING
)
assert result.state == gl.STATE_INVALID
assert result.reason == "news_non_finite"
def price_fixture(
*,
symbol: str = SYMBOL,
direction: str = "CROWDED_SHORT",
verdict: str = "CONFIRMED",
confidence: str = "MEDIUM",
verdict_reason: str = "key_reversal",
as_of: str = "2026-07-14",
schema_version: str = "1.0",
stop_reference: float | None = 1.3720,
include_handoff: bool = True,
week_of: str = "2026-07-06",
) -> dict:
"""Mirrors the real technical-analyst contrarian-confirmation output
shape (check_weekly_price_action.py's output builder)."""
checks = {
"weekly_key_reversal": {
"triggered": True,
"week_of": week_of,
"detail": "new 12-week low reversed",
},
"failed_extreme": {"triggered": False, "week_of": None, "detail": ""},
"failed_breakout": {"triggered": False, "week_of": None, "detail": ""},
}
swing_levels = {
"nearest_swing_high": {"price": 1.4000, "week_of": week_of, "fallback": False},
"nearest_swing_low": {"price": 1.3500, "week_of": week_of, "fallback": False},
"stop_reference": stop_reference,
}
data = {
"symbol": symbol,
"direction": direction,
"mode": "data",
"verdict": verdict,
"confidence": confidence,
"verdict_reason": verdict_reason,
"checks": checks if verdict == "CONFIRMED" else None,
"swing_levels": swing_levels if verdict == "CONFIRMED" else None,
"weekly_bars_used": 60,
"last_completed_week": week_of,
"run_context": {"as_of": as_of, "schema_version": schema_version},
}
if include_handoff:
data["handoff"] = {
"price_action": {
"verdict": verdict,
"confidence": confidence,
"stop_reference": stop_reference,
"report_path": "reports/ta_confirmation_B6_2026-07-14.json",
}
}
return data
def test_normalize_price_action_confirmed_reads_stop_reference_from_handoff() -> None:
data = price_fixture()
result = gl.normalize_price_action(
data, None, symbol=SYMBOL, as_of=AS_OF, max_age_days=7, detector=CONFIRMED_CROWDING
)
assert result.state == gl.STATE_CONFIRMED
assert result.stop_reference == 1.3720
assert result.entry_trigger == "price-action confirmation: key_reversal at week_of=2026-07-06"
def test_normalize_price_action_confirmed_falls_back_to_swing_levels_when_handoff_absent() -> None:
data = price_fixture(include_handoff=False)
result = gl.normalize_price_action(
data, None, symbol=SYMBOL, as_of=AS_OF, max_age_days=7, detector=CONFIRMED_CROWDING
)
assert result.state == gl.STATE_CONFIRMED
assert result.stop_reference == 1.3720
def test_normalize_price_action_confirmed_without_stop_reference_is_invalid() -> None:
"""Fail-closed: CONFIRMED without a usable stop_reference is not
actionable -- a READY_FOR_PLAN without invalidation is unacceptable."""
data = price_fixture(stop_reference=None, include_handoff=False)
data["swing_levels"]["stop_reference"] = None
result = gl.normalize_price_action(
data, None, symbol=SYMBOL, as_of=AS_OF, max_age_days=7, detector=CONFIRMED_CROWDING
)
assert result.state == gl.STATE_INVALID
assert result.reason == "price_action_missing_stop_reference"
# --- Section 6: PR #249 user-review round 2 (P1-A / P1-B) -------------------
def test_normalize_price_action_confirmed_unknown_reason_p1_a_repro() -> None:
"""THE USER'S P1-A REPRO: verdict_reason "BANANA" on a CONFIRMED
price-action report used to reach READY_FOR_PLAN with
entry_trigger="price-action confirmation: BANANA" instead of failing
closed against TA's actual verdict_reason contract."""
data = price_fixture(verdict="CONFIRMED", verdict_reason="BANANA")
result = gl.normalize_price_action(
data, None, symbol=SYMBOL, as_of=AS_OF, max_age_days=7, detector=CONFIRMED_CROWDING
)
assert result.state == gl.STATE_INVALID
assert result.reason == "price_action_unknown_reason"
assert result.entry_trigger is None
def test_normalize_price_action_confirmed_missing_reason_p1_a_repro() -> None:
"""THE USER'S P1-A REPRO (missing variant): a null verdict_reason on a
CONFIRMED report used to reach READY_FOR_PLAN with entry_trigger=null
instead of failing closed."""
data = price_fixture(verdict="CONFIRMED", verdict_reason=None)
result = gl.normalize_price_action(
data, None, symbol=SYMBOL, as_of=AS_OF, max_age_days=7, detector=CONFIRMED_CROWDING
)
assert result.state == gl.STATE_INVALID
assert result.reason == "price_action_malformed"
def test_price_action_confirmed_reasons_matches_technical_analyst_source() -> None:
"""Cross-skill consistency pin (PR #249 user-review round 2, P1-A): the
hardcoded allowlist in gate_logic.PRICE_ACTION_CONFIRMED_REASONS must
match technical-analyst's own weekly_price_action.CHECK_REASON_MAP
values exactly. If TA ever adds a new confirming-signal token, this
test fails loudly here instead of the gate silently rejecting every
real report carrying that new token as price_action_unknown_reason."""
import importlib.util
ta_module_path = (
Path(__file__).resolve().parents[3]
/ "technical-analyst"
/ "scripts"
/ "weekly_price_action.py"
)
assert ta_module_path.exists(), f"technical-analyst module not found at {ta_module_path}"
spec = importlib.util.spec_from_file_location(
"weekly_price_action_for_gate_test", ta_module_path
)
ta_module = importlib.util.module_from_spec(spec)
spec.loader.exec_module(ta_module)
assert set(ta_module.CHECK_REASON_MAP.values()) == gl.PRICE_ACTION_CONFIRMED_REASONS
@pytest.mark.parametrize(
"bad_stop_reference",
[float("inf"), float("-inf"), float("nan"), True, 0, -1.5, "1.372", None],
)
def test_normalize_price_action_confirmed_invalid_stop_reference_p1_b_repro(
bad_stop_reference,
) -> None:
"""THE USER'S P1-B REPRO (and its variants): a non-finite, non-positive,
boolean, or wrong-typed stop_reference must never reach READY_FOR_PLAN's
invalidation_level. `None` is the pre-existing missing-value case
(kept here for a single parametrized sweep); everything else is a
value that was explicitly PROVIDED but is unusable."""
data = price_fixture(stop_reference=bad_stop_reference, include_handoff=False)
data["swing_levels"]["stop_reference"] = bad_stop_reference
result = gl.normalize_price_action(
data, None, symbol=SYMBOL, as_of=AS_OF, max_age_days=7, detector=CONFIRMED_CROWDING
)
assert result.state == gl.STATE_INVALID
if bad_stop_reference is None:
assert result.reason == "price_action_missing_stop_reference"
else:
assert result.reason == "price_action_invalid_stop_reference"
def test_normalize_price_action_confirmed_valid_positive_stop_reference_control() -> None:
"""Control for the P1-B parametrized sweep: a normal, finite, positive
stop_reference must still succeed."""
data = price_fixture(stop_reference=1.372)
result = gl.normalize_price_action(
data, None, symbol=SYMBOL, as_of=AS_OF, max_age_days=7, detector=CONFIRMED_CROWDING
)
assert result.state == gl.STATE_CONFIRMED
assert result.stop_reference == 1.372
def test_normalize_price_action_invalid_stop_reference_in_handoff_does_not_fall_back_to_swing_levels() -> (
None
):
"""An explicit garbage handoff.stop_reference is a real report bug and
must not be silently masked by a valid swing_levels fallback (v4
design decision, pinned)."""
data = price_fixture(stop_reference=1.372) # valid swing_levels.stop_reference
data["handoff"]["price_action"]["stop_reference"] = float("inf")
result = gl.normalize_price_action(
data, None, symbol=SYMBOL, as_of=AS_OF, max_age_days=7, detector=CONFIRMED_CROWDING
)
assert result.state == gl.STATE_INVALID
assert result.reason == "price_action_invalid_stop_reference"
def test_build_gate_result_ready_invariant_catches_a_bypassed_bad_entry_trigger() -> None:
"""Defensive-invariant regression test: even if some future code path
bypassed normalize_price_action's validation and handed
build_gate_result a CONFIRMED price step with no entry_trigger, the
READY_FOR_PLAN invariant assert must catch it rather than silently
emitting an incomplete READY_FOR_PLAN."""
crowding = _crowding("CONF_S")
news = gl.NormalizedInput(kind=gl.STEP_NEWS, state=gl.STATE_CONFIRMED, confidence="HIGH")
price = gl.NormalizedInput(
kind=gl.STEP_PRICE, state=gl.STATE_CONFIRMED, confidence="HIGH", stop_reference=1.37
) # entry_trigger deliberately omitted
with pytest.raises(AssertionError, match="entry_trigger"):
gl.build_gate_result(
symbol=SYMBOL,
crowding=crowding,
news=news,
price=price,
max_detector_age_days=10,
max_report_age_days=7,
as_of=AS_OF,
)
def test_build_gate_result_ready_invariant_catches_a_bypassed_bad_stop_reference() -> None:
crowding = _crowding("CONF_S")
news = gl.NormalizedInput(kind=gl.STEP_NEWS, state=gl.STATE_CONFIRMED, confidence="HIGH")
price = gl.NormalizedInput(
kind=gl.STEP_PRICE,
state=gl.STATE_CONFIRMED,
confidence="HIGH",
entry_trigger="price-action confirmation: key_reversal",
stop_reference=float("inf"),
)
with pytest.raises(AssertionError, match="invalidation_level"):
gl.build_gate_result(
symbol=SYMBOL,
crowding=crowding,
news=news,
price=price,
max_detector_age_days=10,
max_report_age_days=7,
as_of=AS_OF,
)
def test_normalize_price_action_not_confirmed() -> None:
data = price_fixture(verdict="NOT_CONFIRMED", verdict_reason="no_reversal_evidence")
result = gl.normalize_price_action(
data, None, symbol=SYMBOL, as_of=AS_OF, max_age_days=7, detector=CONFIRMED_CROWDING
)
assert result.state == gl.STATE_NOT_CONFIRMED
assert result.reason == "no_reversal_evidence"
def test_normalize_price_action_insufficient_data() -> None:
data = price_fixture(verdict="INSUFFICIENT_DATA", verdict_reason="no_price_source")
result = gl.normalize_price_action(
data, None, symbol=SYMBOL, as_of=AS_OF, max_age_days=7, detector=CONFIRMED_CROWDING
)
assert result.state == gl.STATE_INSUFFICIENT
assert result.reason == "no_price_source"
def test_normalize_price_action_schema_version_read_from_run_context_only() -> None:
"""TA carries schema_version ONLY at run_context.schema_version -- no
top-level key exists. Reading top-level would silently disable the
check (plan §3.2 v2 P1-3): a report with a VALID top-level
schema_version but an UNSUPPORTED run_context.schema_version must
still be rejected -- proving the top-level value is never consulted."""
data = price_fixture()
data["schema_version"] = "1.0" # would pass if (wrongly) read top-level
data["run_context"]["schema_version"] = "2.0" # the actual read location
result = gl.normalize_price_action(
data, None, symbol=SYMBOL, as_of=AS_OF, max_age_days=7, detector=CONFIRMED_CROWDING
)
assert result.state == gl.STATE_INVALID
assert result.reason == "price_action_schema_unsupported"
def test_normalize_price_action_direction_mismatch() -> None:
data = price_fixture(direction="CROWDED_LONG")
result = gl.normalize_price_action(
data, None, symbol=SYMBOL, as_of=AS_OF, max_age_days=7, detector=CONFIRMED_CROWDING
)
assert result.state == gl.STATE_INVALID
assert result.reason == "price_action_direction_mismatch"
def test_normalize_price_action_null_direction_is_insufficient_not_mismatch() -> None:
"""Same fix as news (P3-1): a provided `direction: null` is TA's own
fail-closed exit (no_direction_provided), not a mismatch."""
data = price_fixture(
direction=None,
verdict="INSUFFICIENT_DATA",
verdict_reason="no_direction_provided",
stop_reference=None,
include_handoff=False,
)
result = gl.normalize_price_action(
data, None, symbol=SYMBOL, as_of=AS_OF, max_age_days=7, detector=CONFIRMED_CROWDING
)
assert result.state == gl.STATE_INSUFFICIENT
assert result.reason == "no_direction_provided"
def test_normalize_price_action_null_direction_without_verdict_reason_is_malformed() -> None:
data = price_fixture(
direction=None, verdict_reason=None, stop_reference=None, include_handoff=False
)
result = gl.normalize_price_action(
data, None, symbol=SYMBOL, as_of=AS_OF, max_age_days=7, detector=CONFIRMED_CROWDING
)
assert result.state == gl.STATE_INVALID
assert result.reason == "price_action_malformed"
def test_normalize_price_action_symbol_mismatch() -> None:
data = price_fixture(symbol="D6")
result = gl.normalize_price_action(
data, None, symbol=SYMBOL, as_of=AS_OF, max_age_days=7, detector=CONFIRMED_CROWDING
)
assert result.state == gl.STATE_INVALID
assert result.reason == "price_action_symbol_mismatch"
def test_normalize_price_action_missing_direction_key_is_malformed() -> None:
data = price_fixture()
del data["direction"]
result = gl.normalize_price_action(
data, None, symbol=SYMBOL, as_of=AS_OF, max_age_days=7, detector=CONFIRMED_CROWDING
)
assert result.state == gl.STATE_INVALID
assert result.reason == "price_action_malformed"
def test_normalize_price_action_load_error_parse_error() -> None:
result = gl.normalize_price_action(
None, "parse_error", symbol=SYMBOL, as_of=AS_OF, max_age_days=7, detector=CONFIRMED_CROWDING
)
assert result.state == gl.STATE_INVALID
assert result.reason == "price_action_parse_error"
def test_normalize_price_action_load_error_non_finite() -> None:
result = gl.normalize_price_action(
None, "non_finite", symbol=SYMBOL, as_of=AS_OF, max_age_days=7, detector=CONFIRMED_CROWDING
)
assert result.state == gl.STATE_INVALID
assert result.reason == "price_action_non_finite"
def test_normalize_price_action_unknown_verdict() -> None:
data = price_fixture(verdict="MAYBE")
result = gl.normalize_price_action(
data, None, symbol=SYMBOL, as_of=AS_OF, max_age_days=7, detector=CONFIRMED_CROWDING
)
assert result.state == gl.STATE_INVALID
assert result.reason == "price_action_unknown_verdict"
def test_normalize_price_action_unhashable_verdict_never_crashes() -> None:
"""Mirror of the user's PR #249 P1-1 repro for price-action."""
data = price_fixture(verdict=[])
result = gl.normalize_price_action(
data, None, symbol=SYMBOL, as_of=AS_OF, max_age_days=7, detector=CONFIRMED_CROWDING
)
assert result.state == gl.STATE_INVALID
assert result.reason == "price_action_malformed"
def test_normalize_price_action_unhashable_confidence_never_crashes() -> None:
data = price_fixture(confidence={})
result = gl.normalize_price_action(
data, None, symbol=SYMBOL, as_of=AS_OF, max_age_days=7, detector=CONFIRMED_CROWDING
)
assert result.state == gl.STATE_INVALID
assert result.reason == "price_action_malformed"
def test_normalize_price_action_unknown_confidence_string_fails_closed() -> None:
data = price_fixture(verdict="CONFIRMED", confidence="BANANA")
result = gl.normalize_price_action(
data, None, symbol=SYMBOL, as_of=AS_OF, max_age_days=7, detector=CONFIRMED_CROWDING
)
assert result.state == gl.STATE_INVALID
assert result.reason == "price_action_unknown_confidence"
def test_normalize_price_action_low_confidence_is_accepted_reserved_token() -> None:
data = price_fixture(verdict="CONFIRMED", confidence="LOW")
result = gl.normalize_price_action(
data, None, symbol=SYMBOL, as_of=AS_OF, max_age_days=7, detector=CONFIRMED_CROWDING
)
assert result.state == gl.STATE_CONFIRMED
assert result.confidence == "LOW"
def test_normalize_price_action_non_string_direction_is_malformed_not_mismatch() -> None:
data = price_fixture(direction=123)
result = gl.normalize_price_action(
data, None, symbol=SYMBOL, as_of=AS_OF, max_age_days=7, detector=CONFIRMED_CROWDING
)
assert result.state == gl.STATE_INVALID
assert result.reason == "price_action_malformed"
def test_normalize_price_action_near_stale_warning() -> None:
data = price_fixture(as_of="2026-07-09") # AS_OF 2026-07-15, max 7 -> age 6
result = gl.normalize_price_action(
data, None, symbol=SYMBOL, as_of=AS_OF, max_age_days=7, detector=CONFIRMED_CROWDING
)
assert "price_action_near_stale" in result.warnings
# --- Section 5: gate_confidence / warnings at READY_FOR_PLAN ----------------
def _confirmed_news(confidence: str) -> gl.NormalizedInput:
return gl.NormalizedInput(kind=gl.STEP_NEWS, state=gl.STATE_CONFIRMED, confidence=confidence)
def _confirmed_price(
confidence: str, stop_reference: float = 1.37, entry_trigger: str = "x"
) -> gl.NormalizedInput:
return gl.NormalizedInput(
kind=gl.STEP_PRICE,
state=gl.STATE_CONFIRMED,
confidence=confidence,
stop_reference=stop_reference,
entry_trigger=entry_trigger,
)
@pytest.mark.parametrize(
"news_conf,price_conf,expected",
[
("HIGH", "HIGH", "HIGH"),
("HIGH", "MEDIUM", "MEDIUM"),
("MEDIUM", "HIGH", "MEDIUM"),
("MEDIUM", "MEDIUM", "MEDIUM"),
# LOW is a reserved-but-accepted token (PR #249 P1-2): it ranks
# weakest of all three, below MEDIUM.
("HIGH", "LOW", "LOW"),
("LOW", "HIGH", "LOW"),
("MEDIUM", "LOW", "LOW"),
("LOW", "LOW", "LOW"),
],
)
def test_gate_confidence_is_weakest_link(news_conf, price_conf, expected) -> None:
crowding = _crowding("CONF_S")
news = _confirmed_news(news_conf)
price = _confirmed_price(price_conf)
result = gl.build_gate_result(
symbol=SYMBOL,
crowding=crowding,
news=news,
price=price,
max_detector_age_days=10,
max_report_age_days=7,
as_of=AS_OF,
)
assert result["setup_status"] == "READY_FOR_PLAN"
assert result["gate_confidence"] == expected
def test_price_confidence_medium_warning_at_ready() -> None:
crowding = _crowding("CONF_S")
news = _confirmed_news("HIGH")
price = _confirmed_price("MEDIUM")
result = gl.build_gate_result(
symbol=SYMBOL,
crowding=crowding,
news=news,
price=price,
max_detector_age_days=10,
max_report_age_days=7,
as_of=AS_OF,
)
assert "price_action_confidence_medium" in result["warnings"]
assert "news_confidence_medium" not in result["warnings"]
def test_news_confidence_medium_warning_at_ready() -> None:
crowding = _crowding("CONF_S")
news = _confirmed_news("MEDIUM")
price = _confirmed_price("HIGH")
result = gl.build_gate_result(
symbol=SYMBOL,
crowding=crowding,
news=news,
price=price,
max_detector_age_days=10,
max_report_age_days=7,
as_of=AS_OF,
)
assert "news_confidence_medium" in result["warnings"]
def test_ready_for_plan_populates_entry_trigger_and_invalidation_level() -> None:
crowding = _crowding("CONF_S")
news = _confirmed_news("HIGH")
price = _confirmed_price(
"HIGH",
stop_reference=1.3720,
entry_trigger="price-action confirmation: key_reversal at week_of=2026-07-06",
)
result = gl.build_gate_result(
symbol=SYMBOL,
crowding=crowding,
news=news,
price=price,
max_detector_age_days=10,
max_report_age_days=7,
as_of=AS_OF,
)
assert result["invalidation_level"] == 1.3720
assert (
result["entry_trigger"] == "price-action confirmation: key_reversal at week_of=2026-07-06"
)
assert result["direction"] == "LONG"
def test_inputs_block_shape_for_all_three_steps() -> None:
crowding = _crowding("CONF_S")
news = gl.pending_input(gl.STEP_NEWS)
price = gl.pending_input(gl.STEP_PRICE)
result = gl.build_gate_result(
symbol=SYMBOL,
crowding=crowding,
news=news,
price=price,
max_detector_age_days=10,
max_report_age_days=7,
as_of=AS_OF,
)
assert set(result["inputs"]["crowding"]) == {
"state",
"classification",
"data_date",
"age_days",
"report_path",
}
assert set(result["inputs"]["news_failure"]) == {
"state",
"verdict",
"confidence",
"verdict_reason",
"as_of",
"age_days",
"report_path",
}
assert set(result["inputs"]["price_action"]) == {
"state",
"verdict",
"confidence",
"verdict_reason",
"stop_reference",
"as_of",
"age_days",
"report_path",
}
assert result["setup_status"] == "CROWDED"
assert result["run_context"]["symbol"] == SYMBOL
assert result["run_context"]["schema_version"] == "1.0"
scripts/tests/test_run_contrarian_setup_gate.py
"""Tests for run_contrarian_setup_gate.py -- the CLI wrapper.
Covers: the 4-class JSON-load hardening matrix across all three inputs
(missing file, non-UTF-8 binary, invalid JSON, and top-level wrong-shape --
the last one exercised end-to-end through gate_logic's normalize_*, since
load_json_file() itself only distinguishes unreadable/parse_error), real-
schema fixture end-to-end runs for every setup_status, and Markdown
rendering for every status (never crashes on a null block).
"""
from __future__ import annotations
import json
import sys
from pathlib import Path
from typing import Any
import gate_logic
import pytest
import run_contrarian_setup_gate as cli
SYMBOL = "B6"
AS_OF = "2026-07-15"
# --- Section 1: load_json_file 4-class hardening ----------------------------
def test_load_json_file_missing_file_is_unreadable(tmp_path: Path) -> None:
data, reason = cli.load_json_file(str(tmp_path / "does_not_exist.json"))
assert data is None
assert reason == "unreadable"
def test_load_json_file_directory_is_unreadable(tmp_path: Path) -> None:
data, reason = cli.load_json_file(str(tmp_path))
assert data is None
assert reason == "unreadable"
def test_load_json_file_non_utf8_binary_is_unreadable(tmp_path: Path) -> None:
path = tmp_path / "binary.json"
path.write_bytes(b"\xff\xfe\x00bad")
data, reason = cli.load_json_file(str(path))
assert data is None
assert reason == "unreadable"
def test_load_json_file_invalid_json_is_parse_error(tmp_path: Path) -> None:
path = tmp_path / "bad.json"
path.write_text("{not valid json", encoding="utf-8")
data, reason = cli.load_json_file(str(path))
assert data is None
assert reason == "parse_error"
@pytest.mark.parametrize("raw_text", ["[1, 2, 3]", "null", '"a string"', "42"])
def test_load_json_file_wrong_top_level_shape_loads_but_flows_to_malformed(
tmp_path: Path, raw_text: str
) -> None:
"""Valid JSON but the wrong top-level shape loads fine here -- it is
gate_logic.normalize_*'s job to fail closed to `<input>_malformed`."""
path = tmp_path / "wrong_shape.json"
path.write_text(raw_text, encoding="utf-8")
data, reason = cli.load_json_file(str(path))
assert reason is None
result = gate_logic.normalize_crowding(data, None, symbol=SYMBOL, as_of=AS_OF, max_age_days=10)
assert result.state == gate_logic.STATE_INVALID
assert result.reason == "detector_malformed"
def test_load_json_file_valid_json_succeeds(tmp_path: Path) -> None:
path = tmp_path / "ok.json"
path.write_text(json.dumps({"a": 1}), encoding="utf-8")
data, reason = cli.load_json_file(str(path))
assert reason is None
assert data == {"a": 1}
# --- Section 1b: PR #249 user-review round 3 -- whole-file non-finite scan --
def test_load_json_file_overflow_number_is_non_finite(tmp_path: Path) -> None:
"""A syntactically valid JSON number that overflows to `inf` on parse
(THE USER'S REPRO shape: 1e309) must be caught here, before the data
ever reaches gate_logic."""
path = tmp_path / "overflow.json"
path.write_text('{"classification": 1e309}', encoding="utf-8")
data, reason = cli.load_json_file(str(path))
assert data is None
assert reason == "non_finite"
def test_load_json_file_literal_infinity_is_non_finite(tmp_path: Path) -> None:
"""A bare `Infinity` JSON literal (a non-standard extension json.loads
accepts by default) must also be caught."""
path = tmp_path / "infinity.json"
path.write_text('{"verdict": Infinity}', encoding="utf-8")
data, reason = cli.load_json_file(str(path))
assert data is None
assert reason == "non_finite"
def test_load_json_file_negative_infinity_inside_a_list_is_non_finite(tmp_path: Path) -> None:
"""A non-finite value nested inside a LIST (mirroring a `skipped[]`
entry) must be caught -- the scan is not top-level-keys-only."""
path = tmp_path / "neg_inf_in_list.json"
path.write_text('{"skipped": [{"symbol": "B6", "note": -Infinity}]}', encoding="utf-8")
data, reason = cli.load_json_file(str(path))
assert data is None
assert reason == "non_finite"
def test_load_json_file_nan_literal_nested_deep_is_non_finite(tmp_path: Path) -> None:
"""A NaN literal buried several levels deep (mirroring a nested
run_context field) must be caught -- the scan recurses to any depth."""
path = tmp_path / "nan_deep.json"
path.write_text('{"run_context": {"params": {"nested": {"value": NaN}}}}', encoding="utf-8")
data, reason = cli.load_json_file(str(path))
assert data is None
assert reason == "non_finite"
def test_load_json_file_clean_file_with_ordinary_floats_is_unaffected(tmp_path: Path) -> None:
"""Control: ordinary finite floats (including negative and
fractional) must never trip the non-finite scan."""
path = tmp_path / "clean.json"
path.write_text('{"stop_reference": 1.372, "net_position": -87903.0}', encoding="utf-8")
data, reason = cli.load_json_file(str(path))
assert reason is None
assert data == {"stop_reference": 1.372, "net_position": -87903.0}
def test_contains_non_finite_directly() -> None:
assert cli._contains_non_finite(float("inf")) is True
assert cli._contains_non_finite(float("-inf")) is True
assert cli._contains_non_finite(float("nan")) is True
assert cli._contains_non_finite({"a": {"b": [1, 2, float("inf")]}}) is True
assert cli._contains_non_finite({"a": [1, 2, 3], "b": "text", "c": None, "d": True}) is False
assert cli._contains_non_finite(1.372) is False
assert cli._contains_non_finite(42) is False
# --- Section 1c: PR #249 user-review round 4 -- iterative scan, decoder limits --
def _deep_nested_list(depth: int, leaf: Any) -> Any:
"""Build a genuinely nested Python list `depth` levels deep, with
`leaf` at the bottom -- e.g. depth=3, leaf=1 -> [[[1]]]."""
value = leaf
for _ in range(depth):
value = [value]
return value
def test_contains_non_finite_deep_but_all_finite_returns_false_no_recursion_error() -> None:
"""(a) unit level: THE USER'S ROUND-4 REPRO shape -- a legitimate,
all-finite structure nested 500 levels deep must return False without
raising RecursionError. A plain recursive implementation of this
function is confirmed (empirically, in this environment) to raise
RecursionError somewhere between depth 300 and depth 500."""
deep_finite = _deep_nested_list(500, leaf=1)
assert cli._contains_non_finite(deep_finite) is False
def test_contains_non_finite_deep_with_non_finite_leaf_returns_true_no_recursion_error() -> None:
"""(b) unit level: the same depth, but the leaf is non-finite -- must
still be detected, and still without raising."""
deep_non_finite = _deep_nested_list(500, leaf=float("inf"))
assert cli._contains_non_finite(deep_non_finite) is True
def test_load_json_file_deep_but_finite_nested_field_loads_normally(tmp_path: Path) -> None:
"""(a) load_json_file level: a real detector-shaped file with an
extra, unused, all-finite field nested 500 levels deep must load
completely normally -- reason is None, exactly as if the field were
absent."""
fixture = _detector_fixture()
fixture["_deep_unused_field"] = _deep_nested_list(500, leaf=1)
path = tmp_path / "deep_finite.json"
path.write_text(json.dumps(fixture), encoding="utf-8")
data, reason = cli.load_json_file(str(path))
assert reason is None
assert data["markets"][0]["classification"] == "CROWDED_SHORT"
def test_load_json_file_deep_with_non_finite_leaf_is_non_finite(tmp_path: Path) -> None:
"""(b) load_json_file level: same shape, but the deeply-nested leaf
is non-finite -- must still be caught as `non_finite`."""
fixture = _detector_fixture()
fixture["_deep_unused_field"] = _deep_nested_list(500, leaf=float("inf"))
path = tmp_path / "deep_non_finite.json"
path.write_text(json.dumps(fixture), encoding="utf-8")
data, reason = cli.load_json_file(str(path))
assert data is None
assert reason == "non_finite"
def test_load_json_file_extreme_nesting_beyond_decoder_limit_is_parse_error(tmp_path: Path) -> None:
"""(c) A depth far beyond what even the JSON decoder itself can
handle must fail closed as `parse_error`, not crash with an uncaught
RecursionError. Verified empirically in this environment: json.loads
on a `[[[...1...]]]` string handles depth 100,000 fine but raises
RecursionError around depth 200,000 ("Stack overflow ... while
decoding a JSON array") -- 250,000 is comfortably past that
threshold. RecursionError is NOT a subclass of json.JSONDecodeError
or ValueError, so it must be caught explicitly."""
depth = 250_000
raw_text = "[" * depth + "1" + "]" * depth
path = tmp_path / "extreme_nesting.json"
path.write_text(raw_text, encoding="utf-8")
data, reason = cli.load_json_file(str(path))
assert data is None
assert reason == "parse_error"
def test_cli_deep_but_finite_unused_field_produces_normal_result(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
"""(a) End-to-end: a detector report with an unused, all-finite,
deeply-nested field must produce the EXACT SAME result as the
ordinary fixture without that field -- CROWDED, not a rejection. The
non-finite scan must have zero false-positive cost on legitimate
input, however deeply nested."""
fixture = _detector_fixture()
fixture["_deep_unused_field"] = _deep_nested_list(500, leaf=1)
detector_path = tmp_path / "detector.json"
detector_path.write_text(json.dumps(fixture), encoding="utf-8")
exit_code = _run_cli(
monkeypatch,
tmp_path,
["--symbol", SYMBOL, "--detector-json", str(detector_path), "--as-of", AS_OF],
)
assert exit_code == 0
result = json.loads((tmp_path / f"contrarian_setup_gate_{SYMBOL}_{AS_OF}.json").read_text())
assert result["setup_status"] == "CROWDED"
assert result["direction"] == "LONG"
def test_cli_deep_with_non_finite_leaf_produces_non_finite_report(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
"""(b) End-to-end: the same shape, but the deep leaf is non-finite --
exit 0, both reports written, reason detector_non_finite."""
fixture = _detector_fixture()
fixture["_deep_unused_field"] = _deep_nested_list(500, leaf=float("inf"))
detector_path = tmp_path / "detector.json"
detector_path.write_text(json.dumps(fixture), encoding="utf-8")
exit_code = _run_cli(
monkeypatch,
tmp_path,
["--symbol", SYMBOL, "--detector-json", str(detector_path), "--as-of", AS_OF],
)
assert exit_code == 0
result = json.loads((tmp_path / f"contrarian_setup_gate_{SYMBOL}_{AS_OF}.json").read_text())
assert result["setup_status"] == "INSUFFICIENT_EVIDENCE"
assert any(item["reason"] == "detector_non_finite" for item in result["missing_confirmations"])
def test_cli_extreme_nesting_beyond_decoder_limit_produces_parse_error_report(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
"""(c) End-to-end: nesting far beyond the decoder's own limit must
exit 0 with a full report naming detector_parse_error, never crash."""
depth = 250_000
raw_text = "[" * depth + "1" + "]" * depth
detector_path = tmp_path / "detector.json"
detector_path.write_text(raw_text, encoding="utf-8")
exit_code = _run_cli(
monkeypatch,
tmp_path,
["--symbol", SYMBOL, "--detector-json", str(detector_path), "--as-of", AS_OF],
)
assert exit_code == 0
json_path = tmp_path / f"contrarian_setup_gate_{SYMBOL}_{AS_OF}.json"
md_path = tmp_path / f"contrarian_setup_gate_{SYMBOL}_{AS_OF}.md"
assert json_path.exists()
assert md_path.exists()
result = json.loads(json_path.read_text())
assert result["setup_status"] == "INSUFFICIENT_EVIDENCE"
assert any(item["reason"] == "detector_parse_error" for item in result["missing_confirmations"])
# --- Section 2: fixture builders (real-schema shapes) -----------------------
def _detector_fixture(*, symbol=SYMBOL, classification="CROWDED_SHORT", data_date="2026-07-07"):
return {
"schema_version": "1.0",
"skill": "cot-contrarian-detector",
"run_context": {"schema_version": "1.0", "as_of": "2026-07-12", "data_date": data_date},
"markets": [
{
"symbol": symbol,
"status": "ok",
"data_date": data_date,
"classification": classification,
}
],
"skipped": [],
}
def _news_fixture(
*,
symbol=SYMBOL,
direction="CROWDED_SHORT",
verdict="CONFIRMED",
confidence="HIGH",
as_of="2026-07-13",
):
return {
"schema_version": "1.0",
"skill": "news-reaction-failure-analyzer",
"symbol": symbol,
"direction": direction,
"expected_direction": "BULLISH",
"actual_reaction": "NO_REACTION",
"verdict": verdict,
"confidence": confidence,
"verdict_reason": "no_significant_drift" if verdict != "CONFIRMED" else None,
"relevant_events_used": 3,
"aggregate": {"mean_z3": 0.1, "drift_stat": 0.2, "responded_ratio": 0.33},
"evidence": [],
"clusters": [],
"dropped_events": [],
"run_context": {"as_of": as_of},
}
def _price_fixture(
*,
symbol=SYMBOL,
direction="CROWDED_SHORT",
verdict="CONFIRMED",
confidence="HIGH",
as_of="2026-07-14",
):
checks = {
"weekly_key_reversal": {"triggered": True, "week_of": "2026-07-06", "detail": ""},
"failed_extreme": {"triggered": False, "week_of": None, "detail": ""},
"failed_breakout": {"triggered": False, "week_of": None, "detail": ""},
}
return {
"symbol": symbol,
"direction": direction,
"mode": "data",
"verdict": verdict,
"confidence": confidence,
"verdict_reason": "key_reversal" if verdict == "CONFIRMED" else "no_reversal_evidence",
"checks": checks if verdict == "CONFIRMED" else None,
"swing_levels": {"stop_reference": 1.3720} if verdict == "CONFIRMED" else None,
"handoff": {
"price_action": {
"verdict": verdict,
"confidence": confidence,
"stop_reference": 1.3720 if verdict == "CONFIRMED" else None,
}
},
"run_context": {"as_of": as_of, "schema_version": "1.0"},
}
def _run_cli(monkeypatch, tmp_path: Path, argv: list[str]) -> int:
monkeypatch.setattr(
sys, "argv", ["run_contrarian_setup_gate.py", *argv, "--output-dir", str(tmp_path)]
)
return cli.main()
def _write(tmp_path: Path, name: str, payload) -> str:
path = tmp_path / name
path.write_text(json.dumps(payload), encoding="utf-8")
return str(path)
# --- Section 3: end-to-end scenarios -----------------------------------------
def test_cli_detector_only_crowded(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
detector_path = _write(tmp_path, "detector.json", _detector_fixture())
exit_code = _run_cli(
monkeypatch,
tmp_path,
["--symbol", SYMBOL, "--detector-json", detector_path, "--as-of", AS_OF],
)
assert exit_code == 0
result = json.loads((tmp_path / f"contrarian_setup_gate_{SYMBOL}_{AS_OF}.json").read_text())
assert result["setup_status"] == "CROWDED"
assert result["direction"] == "LONG"
assert (tmp_path / f"contrarian_setup_gate_{SYMBOL}_{AS_OF}.md").exists()
def test_cli_full_trio_ready_for_plan(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
detector_path = _write(tmp_path, "detector.json", _detector_fixture())
news_path = _write(tmp_path, "news.json", _news_fixture(verdict="CONFIRMED"))
price_path = _write(tmp_path, "price.json", _price_fixture(verdict="CONFIRMED"))
exit_code = _run_cli(
monkeypatch,
tmp_path,
[
"--symbol",
SYMBOL,
"--detector-json",
detector_path,
"--news-json",
news_path,
"--price-action-json",
price_path,
"--as-of",
AS_OF,
],
)
assert exit_code == 0
result = json.loads((tmp_path / f"contrarian_setup_gate_{SYMBOL}_{AS_OF}.json").read_text())
assert result["setup_status"] == "READY_FOR_PLAN"
assert result["direction"] == "LONG"
assert result["gate_confidence"] == "HIGH"
assert result["invalidation_level"] == 1.3720
assert result["entry_trigger"] is not None
def test_cli_news_not_confirmed_rejects(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
detector_path = _write(tmp_path, "detector.json", _detector_fixture())
news_path = _write(tmp_path, "news.json", _news_fixture(verdict="NOT_CONFIRMED"))
exit_code = _run_cli(
monkeypatch,
tmp_path,
[
"--symbol",
SYMBOL,
"--detector-json",
detector_path,
"--news-json",
news_path,
"--as-of",
AS_OF,
],
)
assert exit_code == 0
result = json.loads((tmp_path / f"contrarian_setup_gate_{SYMBOL}_{AS_OF}.json").read_text())
assert result["setup_status"] == "REJECTED"
assert result["missing_confirmations"][0]["step"] == "news_failure"
def test_cli_binary_detector_file_never_crashes(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
"""Binary/corrupt detector JSON -> INSUFFICIENT_EVIDENCE, exit 0, a
report is written, and stdout carries no Python traceback."""
detector_path = tmp_path / "detector.json"
detector_path.write_bytes(b"\xff\xfe\x00bad")
exit_code = _run_cli(
monkeypatch,
tmp_path,
["--symbol", SYMBOL, "--detector-json", str(detector_path), "--as-of", AS_OF],
)
assert exit_code == 0
result = json.loads((tmp_path / f"contrarian_setup_gate_{SYMBOL}_{AS_OF}.json").read_text())
assert result["setup_status"] == "INSUFFICIENT_EVIDENCE"
assert result["inputs"]["crowding"]["state"] == "INVALID"
assert result["missing_confirmations"][0]["reason"] == "detector_unreadable"
def test_cli_missing_detector_file_never_crashes(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
exit_code = _run_cli(
monkeypatch,
tmp_path,
["--symbol", SYMBOL, "--detector-json", str(tmp_path / "nope.json"), "--as-of", AS_OF],
)
assert exit_code == 0
result = json.loads((tmp_path / f"contrarian_setup_gate_{SYMBOL}_{AS_OF}.json").read_text())
assert result["setup_status"] == "INSUFFICIENT_EVIDENCE"
def test_cli_malformed_news_json_top_level_list_never_crashes(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
detector_path = _write(tmp_path, "detector.json", _detector_fixture())
news_path = _write(tmp_path, "news.json", [1, 2, 3])
exit_code = _run_cli(
monkeypatch,
tmp_path,
[
"--symbol",
SYMBOL,
"--detector-json",
detector_path,
"--news-json",
news_path,
"--as-of",
AS_OF,
],
)
assert exit_code == 0
result = json.loads((tmp_path / f"contrarian_setup_gate_{SYMBOL}_{AS_OF}.json").read_text())
assert result["setup_status"] == "INSUFFICIENT_EVIDENCE"
assert result["inputs"]["news_failure"]["state"] == "INVALID"
def test_cli_news_verdict_list_never_crashes(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
"""THE USER'S PR #249 P1-1 REPRO, end-to-end through the CLI:
`verdict: []` in a provided news report used to raise an uncaught
TypeError (unhashable set-membership check) instead of the exit-0 +
report contract every other degraded input honors."""
detector_path = _write(tmp_path, "detector.json", _detector_fixture())
news_data = _news_fixture(verdict="CONFIRMED")
news_data["verdict"] = []
news_path = _write(tmp_path, "news.json", news_data)
exit_code = _run_cli(
monkeypatch,
tmp_path,
[
"--symbol",
SYMBOL,
"--detector-json",
detector_path,
"--news-json",
news_path,
"--as-of",
AS_OF,
],
)
assert exit_code == 0
result = json.loads((tmp_path / f"contrarian_setup_gate_{SYMBOL}_{AS_OF}.json").read_text())
assert result["setup_status"] == "INSUFFICIENT_EVIDENCE"
assert result["inputs"]["news_failure"]["state"] == "INVALID"
assert result["missing_confirmations"][0]["reason"] == "news_malformed"
def test_cli_news_confidence_dict_never_crashes(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
"""THE USER'S PR #249 P1-1 REPRO, end-to-end: `confidence: {}` used to
raise an uncaught TypeError deep inside gate_confidence computation."""
detector_path = _write(tmp_path, "detector.json", _detector_fixture())
news_data = _news_fixture(verdict="CONFIRMED")
news_data["confidence"] = {}
news_path = _write(tmp_path, "news.json", news_data)
exit_code = _run_cli(
monkeypatch,
tmp_path,
[
"--symbol",
SYMBOL,
"--detector-json",
detector_path,
"--news-json",
news_path,
"--as-of",
AS_OF,
],
)
assert exit_code == 0
result = json.loads((tmp_path / f"contrarian_setup_gate_{SYMBOL}_{AS_OF}.json").read_text())
assert result["setup_status"] == "INSUFFICIENT_EVIDENCE"
assert result["inputs"]["news_failure"]["state"] == "INVALID"
assert result["missing_confirmations"][0]["reason"] == "news_malformed"
def test_cli_detector_classification_int_never_crashes(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
"""Mirror of the user's PR #249 P1-1 repro for the detector's own
`classification` field: a wrong-typed value used to raise TypeError
in the `classification not in FADE_DIRECTION` membership check."""
detector_data = _detector_fixture()
detector_data["markets"][0]["classification"] = 123
detector_path = _write(tmp_path, "detector.json", detector_data)
exit_code = _run_cli(
monkeypatch,
tmp_path,
["--symbol", SYMBOL, "--detector-json", detector_path, "--as-of", AS_OF],
)
assert exit_code == 0
result = json.loads((tmp_path / f"contrarian_setup_gate_{SYMBOL}_{AS_OF}.json").read_text())
assert result["setup_status"] == "INSUFFICIENT_EVIDENCE"
assert result["inputs"]["crowding"]["state"] == "INVALID"
assert result["missing_confirmations"][0]["reason"] == "detector_unknown_classification"
def test_cli_news_confidence_banana_fails_closed_not_passthrough(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
"""PR #249 P1-2 REPRO, end-to-end: an unknown confidence string used
to reach `gate_confidence` in the output verbatim instead of being
rejected as INVALID with a named reason."""
detector_path = _write(tmp_path, "detector.json", _detector_fixture())
price_path = _write(tmp_path, "price.json", _price_fixture(verdict="CONFIRMED"))
news_data = _news_fixture(verdict="CONFIRMED")
news_data["confidence"] = "BANANA"
news_path = _write(tmp_path, "news.json", news_data)
exit_code = _run_cli(
monkeypatch,
tmp_path,
[
"--symbol",
SYMBOL,
"--detector-json",
detector_path,
"--news-json",
news_path,
"--price-action-json",
price_path,
"--as-of",
AS_OF,
],
)
assert exit_code == 0
result = json.loads((tmp_path / f"contrarian_setup_gate_{SYMBOL}_{AS_OF}.json").read_text())
assert result["setup_status"] == "INSUFFICIENT_EVIDENCE"
assert result["gate_confidence"] is None
assert result["inputs"]["news_failure"]["state"] == "INVALID"
assert result["missing_confirmations"][0]["reason"] == "news_unknown_confidence"
def test_cli_price_verdict_reason_banana_p1_a_repro(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
"""THE USER'S P1-A REPRO, end-to-end: verdict_reason "BANANA" on a
CONFIRMED price-action report used to reach READY_FOR_PLAN with
entry_trigger="price-action confirmation: BANANA"."""
detector_path = _write(tmp_path, "detector.json", _detector_fixture())
news_path = _write(tmp_path, "news.json", _news_fixture(verdict="CONFIRMED"))
price_data = _price_fixture(verdict="CONFIRMED")
price_data["verdict_reason"] = "BANANA"
price_path = _write(tmp_path, "price.json", price_data)
exit_code = _run_cli(
monkeypatch,
tmp_path,
[
"--symbol",
SYMBOL,
"--detector-json",
detector_path,
"--news-json",
news_path,
"--price-action-json",
price_path,
"--as-of",
AS_OF,
],
)
assert exit_code == 0
result = json.loads((tmp_path / f"contrarian_setup_gate_{SYMBOL}_{AS_OF}.json").read_text())
assert result["setup_status"] == "INSUFFICIENT_EVIDENCE"
assert result["entry_trigger"] is None
assert result["inputs"]["price_action"]["state"] == "INVALID"
assert result["missing_confirmations"][0]["reason"] == "price_action_unknown_reason"
def test_cli_price_verdict_reason_missing_p1_a_repro(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
"""THE USER'S P1-A REPRO (missing variant), end-to-end: a null
verdict_reason on a CONFIRMED report used to reach READY_FOR_PLAN with
entry_trigger=null."""
detector_path = _write(tmp_path, "detector.json", _detector_fixture())
news_path = _write(tmp_path, "news.json", _news_fixture(verdict="CONFIRMED"))
price_data = _price_fixture(verdict="CONFIRMED")
price_data["verdict_reason"] = None
price_path = _write(tmp_path, "price.json", price_data)
exit_code = _run_cli(
monkeypatch,
tmp_path,
[
"--symbol",
SYMBOL,
"--detector-json",
detector_path,
"--news-json",
news_path,
"--price-action-json",
price_path,
"--as-of",
AS_OF,
],
)
assert exit_code == 0
result = json.loads((tmp_path / f"contrarian_setup_gate_{SYMBOL}_{AS_OF}.json").read_text())
assert result["setup_status"] == "INSUFFICIENT_EVIDENCE"
assert result["missing_confirmations"][0]["reason"] == "price_action_malformed"
def test_cli_price_stop_reference_overflow_to_infinity_p1_b_repro(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
"""THE USER'S P1-B REPRO, end-to-end: `stop_reference: 1e309` is a
syntactically valid JSON number that overflows to Python's `inf` on
parse. This used to reach READY_FOR_PLAN with
"invalidation_level": Infinity in the written file -- not valid
standard JSON, and unusable as a stop level for a downstream
position-sizing skill. As of the round-3 fix, the CLI's whole-file
non-finite scan now catches this BEFORE gate_logic's stop_reference-
specific validation ever runs, so the reason is `price_action_non_finite`
(not `price_action_invalid_stop_reference`, which is still reachable
for finite-but-otherwise-bad values like 0, negative, or bool)."""
detector_path = _write(tmp_path, "detector.json", _detector_fixture())
news_path = _write(tmp_path, "news.json", _news_fixture(verdict="CONFIRMED"))
price_data = _price_fixture(verdict="CONFIRMED")
price_data["swing_levels"]["stop_reference"] = 1e309
price_data["handoff"]["price_action"]["stop_reference"] = 1e309
price_path = _write(tmp_path, "price.json", price_data)
exit_code = _run_cli(
monkeypatch,
tmp_path,
[
"--symbol",
SYMBOL,
"--detector-json",
detector_path,
"--news-json",
news_path,
"--price-action-json",
price_path,
"--as-of",
AS_OF,
],
)
assert exit_code == 0
output_path = tmp_path / f"contrarian_setup_gate_{SYMBOL}_{AS_OF}.json"
output_text = output_path.read_text()
assert "Infinity" not in output_text # never emitted, even indirectly
result = json.loads(output_text) # standard-JSON-compliant: no NaN/Infinity tokens to parse
assert result["setup_status"] == "INSUFFICIENT_EVIDENCE"
assert result["invalidation_level"] is None
assert result["missing_confirmations"][0]["reason"] == "price_action_non_finite"
def test_cli_price_stop_reference_nan_literal_p1_b_repro(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
"""A bare `NaN` JSON literal (a non-standard extension `json.loads`
accepts by default) must also fail closed, not just numeric overflow.
As of the round-3 fix, this is caught by the CLI's whole-file scan
(`price_action_non_finite`) before gate_logic's own stop_reference
validation would otherwise have run."""
detector_path = _write(tmp_path, "detector.json", _detector_fixture())
news_path = _write(tmp_path, "news.json", _news_fixture(verdict="CONFIRMED"))
price_data = _price_fixture(verdict="CONFIRMED")
price_path = tmp_path / "price_nan.json"
raw_text = json.dumps(price_data).replace("1.372", "NaN")
price_path.write_text(raw_text, encoding="utf-8")
exit_code = _run_cli(
monkeypatch,
tmp_path,
[
"--symbol",
SYMBOL,
"--detector-json",
detector_path,
"--news-json",
news_path,
"--price-action-json",
str(price_path),
"--as-of",
AS_OF,
],
)
assert exit_code == 0
result = json.loads((tmp_path / f"contrarian_setup_gate_{SYMBOL}_{AS_OF}.json").read_text())
assert result["setup_status"] == "INSUFFICIENT_EVIDENCE"
assert result["missing_confirmations"][0]["reason"] == "price_action_non_finite"
def test_cli_price_stop_reference_bool_true_p1_b_repro(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
"""`isinstance(True, int)` is True in Python -- a bare `true` JSON
literal for stop_reference must not silently pass as 1.0. Unlike the
overflow/NaN repros above, `bool` is never a `float`, so the CLI's
whole-file non-finite scan does not intercept it -- this still
reaches gate_logic's own `price_action_invalid_stop_reference` check,
confirming that check remains reachable for finite-but-wrong-typed
values after the round-3 fix."""
detector_path = _write(tmp_path, "detector.json", _detector_fixture())
news_path = _write(tmp_path, "news.json", _news_fixture(verdict="CONFIRMED"))
price_data = _price_fixture(verdict="CONFIRMED")
price_data["swing_levels"]["stop_reference"] = True
price_data["handoff"]["price_action"]["stop_reference"] = True
price_path = _write(tmp_path, "price.json", price_data)
exit_code = _run_cli(
monkeypatch,
tmp_path,
[
"--symbol",
SYMBOL,
"--detector-json",
detector_path,
"--news-json",
news_path,
"--price-action-json",
price_path,
"--as-of",
AS_OF,
],
)
assert exit_code == 0
result = json.loads((tmp_path / f"contrarian_setup_gate_{SYMBOL}_{AS_OF}.json").read_text())
assert result["setup_status"] == "INSUFFICIENT_EVIDENCE"
assert result["missing_confirmations"][0]["reason"] == "price_action_invalid_stop_reference"
def test_generate_json_report_allow_nan_false_raises_on_residual_non_finite_value(
tmp_path: Path,
) -> None:
"""Writer-level defense-in-depth test (PR #249 user-review round 2,
P1-B): if a non-finite float ever slipped past gate_logic's own
validation (a future regression), the CLI's JSON writer must raise
loudly via allow_nan=False rather than silently emitting non-standard
JSON."""
bad_result = {"setup_status": "READY_FOR_PLAN", "invalidation_level": float("inf")}
output_path = tmp_path / "bad.json"
with pytest.raises(ValueError):
cli.generate_json_report(bad_result, output_path)
# --- Section 4: PR #249 user-review round 3 -- non-finite anywhere, end-to-end --
def _assert_clean_non_finite_report(
tmp_path: Path, expected_status: str, expected_reason: str
) -> None:
"""Shared assertions for the (a)-(d) repros: exit 0, BOTH reports
written (JSON and MD), the expected status/reason, and no traceback
left in the JSON report's content (a crash would have produced no
file, or a partial one -- reading it back via json.loads is itself
proof the file is valid, complete JSON)."""
json_path = tmp_path / f"contrarian_setup_gate_{SYMBOL}_{AS_OF}.json"
md_path = tmp_path / f"contrarian_setup_gate_{SYMBOL}_{AS_OF}.md"
assert json_path.exists(), "JSON report was not written -- the CLI must never partially fail"
assert md_path.exists(), "Markdown report was not written -- the CLI must never partially fail"
result = json.loads(json_path.read_text()) # raises if the file isn't valid, complete JSON
assert result["setup_status"] == expected_status
assert any(item["reason"] == expected_reason for item in result["missing_confirmations"])
def test_cli_detector_classification_overflow_p1_round3_repro(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
"""(a) THE USER'S ROUND-3 REPRO: detector classification: 1e309.
Before this fix, normalization correctly marked crowding INVALID, but
then the raw `inf` value echoed into the audit block crashed the
writer's allow_nan=False -> exit 1, no/partial report."""
detector_data = _detector_fixture()
detector_data["markets"][0]["classification"] = 1e309
detector_path = tmp_path / "detector.json"
detector_path.write_text(json.dumps(detector_data), encoding="utf-8")
exit_code = _run_cli(
monkeypatch,
tmp_path,
["--symbol", SYMBOL, "--detector-json", str(detector_path), "--as-of", AS_OF],
)
assert exit_code == 0
_assert_clean_non_finite_report(tmp_path, "INSUFFICIENT_EVIDENCE", "detector_non_finite")
def test_cli_news_verdict_overflow_p1_round3_repro(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
"""(b) THE USER'S ROUND-3 REPRO variant: news verdict: 1e309."""
detector_path = _write(tmp_path, "detector.json", _detector_fixture())
news_data = _news_fixture(verdict="CONFIRMED")
news_data["verdict"] = 1e309
news_path = tmp_path / "news.json"
news_path.write_text(json.dumps(news_data), encoding="utf-8")
exit_code = _run_cli(
monkeypatch,
tmp_path,
[
"--symbol",
SYMBOL,
"--detector-json",
detector_path,
"--news-json",
str(news_path),
"--as-of",
AS_OF,
],
)
assert exit_code == 0
_assert_clean_non_finite_report(tmp_path, "INSUFFICIENT_EVIDENCE", "news_non_finite")
def test_cli_nan_literal_nested_deep_in_run_context_p1_round3_repro(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
"""(c) A bare NaN literal nested deep inside run_context -- a field
the gate doesn't even read -- must still reject the whole file."""
detector_data = _detector_fixture()
detector_path = tmp_path / "detector.json"
raw_text = json.dumps(detector_data)
# Inject a deeply-nested NaN into run_context.params without disturbing
# the rest of the structure.
raw_text = raw_text.replace(
'"data_date": "2026-07-07"}',
'"data_date": "2026-07-07", "params": {"nested": {"value": NaN}}}',
1,
)
detector_path.write_text(raw_text, encoding="utf-8")
exit_code = _run_cli(
monkeypatch,
tmp_path,
["--symbol", SYMBOL, "--detector-json", str(detector_path), "--as-of", AS_OF],
)
assert exit_code == 0
_assert_clean_non_finite_report(tmp_path, "INSUFFICIENT_EVIDENCE", "detector_non_finite")
def test_cli_negative_infinity_inside_skipped_entry_p1_round3_repro(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
"""(d) -Infinity inside a skipped[] entry -- again, a field the gate's
decision logic doesn't read for a symbol that IS present in
markets[] -- must still reject the whole file (whole-file scan, not
field-scoped)."""
detector_data = _detector_fixture()
detector_data["skipped"] = [{"symbol": "ZZ", "note": float("-inf")}]
detector_path = tmp_path / "detector.json"
# json.dumps (default allow_nan=True) renders float('-inf') as the
# bare `-Infinity` token, which json.loads reads back as -inf.
raw_text = json.dumps(detector_data)
assert "-Infinity" in raw_text
detector_path.write_text(raw_text, encoding="utf-8")
exit_code = _run_cli(
monkeypatch,
tmp_path,
["--symbol", SYMBOL, "--detector-json", str(detector_path), "--as-of", AS_OF],
)
assert exit_code == 0
_assert_clean_non_finite_report(tmp_path, "INSUFFICIENT_EVIDENCE", "detector_non_finite")
def test_cli_clean_file_control_unaffected_by_non_finite_scan(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
"""(e) Control: an entirely ordinary, finite-numbers-only trio must
still reach READY_FOR_PLAN exactly as before -- the new scan must not
have any false-positive cost on legitimate reports."""
detector_path = _write(tmp_path, "detector.json", _detector_fixture())
news_path = _write(tmp_path, "news.json", _news_fixture(verdict="CONFIRMED"))
price_path = _write(tmp_path, "price.json", _price_fixture(verdict="CONFIRMED"))
exit_code = _run_cli(
monkeypatch,
tmp_path,
[
"--symbol",
SYMBOL,
"--detector-json",
detector_path,
"--news-json",
news_path,
"--price-action-json",
price_path,
"--as-of",
AS_OF,
],
)
assert exit_code == 0
result = json.loads((tmp_path / f"contrarian_setup_gate_{SYMBOL}_{AS_OF}.json").read_text())
assert result["setup_status"] == "READY_FOR_PLAN"
def test_cli_format_json_only_skips_markdown(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
detector_path = _write(tmp_path, "detector.json", _detector_fixture())
_run_cli(
monkeypatch,
tmp_path,
[
"--symbol",
SYMBOL,
"--detector-json",
detector_path,
"--as-of",
AS_OF,
"--format",
"json",
],
)
assert (tmp_path / f"contrarian_setup_gate_{SYMBOL}_{AS_OF}.json").exists()
assert not (tmp_path / f"contrarian_setup_gate_{SYMBOL}_{AS_OF}.md").exists()
def test_cli_invalid_as_of_is_argparse_error(monkeypatch: pytest.MonkeyPatch, capsys) -> None:
monkeypatch.setattr(
sys,
"argv",
[
"run_contrarian_setup_gate.py",
"--symbol",
SYMBOL,
"--detector-json",
"x.json",
"--as-of",
"not-a-date",
],
)
with pytest.raises(SystemExit) as exc_info:
cli.parse_arguments()
assert exc_info.value.code == 2
# --- Section 4: Markdown rendering for every status --------------------------
def _result_for_status(status: str) -> dict:
crowding = gate_logic.NormalizedInput(
kind=gate_logic.STEP_CROWDING,
state=gate_logic.STATE_CONFIRMED
if status != "INSUFFICIENT_EVIDENCE"
else gate_logic.STATE_INVALID,
classification="CROWDED_SHORT" if status != "INSUFFICIENT_EVIDENCE" else None,
direction="LONG" if status != "INSUFFICIENT_EVIDENCE" else None,
reason=None if status != "INSUFFICIENT_EVIDENCE" else "detector_json_stale",
)
if status == "REJECTED":
news = gate_logic.NormalizedInput(
kind=gate_logic.STEP_NEWS,
state=gate_logic.STATE_NOT_CONFIRMED,
reason="no_reversal_evidence",
)
price = gate_logic.pending_input(gate_logic.STEP_PRICE)
elif status == "CROWDED":
news = gate_logic.pending_input(gate_logic.STEP_NEWS)
price = gate_logic.pending_input(gate_logic.STEP_PRICE)
elif status == "WATCHING_PRICE":
news = gate_logic.NormalizedInput(
kind=gate_logic.STEP_NEWS, state=gate_logic.STATE_CONFIRMED, confidence="HIGH"
)
price = gate_logic.pending_input(gate_logic.STEP_PRICE)
elif status == "READY_FOR_PLAN":
news = gate_logic.NormalizedInput(
kind=gate_logic.STEP_NEWS, state=gate_logic.STATE_CONFIRMED, confidence="HIGH"
)
price = gate_logic.NormalizedInput(
kind=gate_logic.STEP_PRICE,
state=gate_logic.STATE_CONFIRMED,
confidence="MEDIUM",
stop_reference=1.372,
entry_trigger="price-action confirmation: key_reversal at week_of=2026-07-06",
)
else: # INSUFFICIENT_EVIDENCE
news = gate_logic.pending_input(gate_logic.STEP_NEWS)
price = gate_logic.pending_input(gate_logic.STEP_PRICE)
return gate_logic.build_gate_result(
symbol=SYMBOL,
crowding=crowding,
news=news,
price=price,
max_detector_age_days=10,
max_report_age_days=7,
as_of=AS_OF,
)
@pytest.mark.parametrize(
"status", ["READY_FOR_PLAN", "WATCHING_PRICE", "CROWDED", "REJECTED", "INSUFFICIENT_EVIDENCE"]
)
def test_markdown_report_renders_for_every_status(tmp_path: Path, status: str) -> None:
result = _result_for_status(status)
assert result["setup_status"] == status
md_path = tmp_path / "report.md"
cli.generate_markdown_report(result, md_path) # must not raise on any null block
text = md_path.read_text(encoding="utf-8")
assert status in text
assert result["symbol"] in text
def test_markdown_report_ready_for_plan_shows_warning() -> None:
result = _result_for_status("READY_FOR_PLAN")
assert "price_action_confidence_medium" in result["warnings"]
SKILL.md
---
name: contrarian-setup-gate
description: Synthesize the three Jason Shapiro contrarian-pipeline verdicts (COT crowding, news-reaction failure, weekly price-action confirmation) into one actionable setup_status via a fail-closed precedence state machine. Pure, offline synthesis -- no network, no API keys, no computation beyond validation and precedence.
---
# Contrarian Setup Gate
## Overview
Synthesize the outputs of Jason Shapiro's 3-step contrarian process into one actionable state. cot-contrarian-detector (step 1) flags crowded positioning, news-reaction-failure-analyzer (step 2) tests whether the market failed to react to news favorable to the crowd, and technical-analyst's contrarian-confirmation mode (step 3) confirms weekly price-action evidence of a reversal. This gate reads those three report JSONs and applies an explicit, exhaustively-tested precedence rule set to produce one `setup_status`, with a fail-closed reason attached to every input that could not be confirmed.
The gate does no fetching, no API calls, and no computation beyond validating and combining the three inputs it is given. It is the pipeline's synthesis center, not a data source.
## When to Use
- After running cot-contrarian-detector (always required -- this is the pipeline's entry point)
- Mid-pipeline, with only the detector report, to see the CROWDED state and what steps remain
- After running news-reaction-failure-analyzer, to see whether the setup advances to WATCHING_PRICE or is REJECTED
- After running technical-analyst's contrarian-confirmation mode, to see whether the setup reaches READY_FOR_PLAN
- Before handing a symbol's direction and stop level to a position-sizing skill
## Prerequisites
- Python 3.9+
- No API keys -- this skill is fully offline
- A cot-contrarian-detector JSON report for the symbol under evaluation (required)
- Optionally, a news-reaction-failure-analyzer JSON report for the same symbol (step 2)
- Optionally, a technical-analyst contrarian-confirmation JSON report for the same symbol (step 3)
## Workflow
### Step 1: Run the Gate
```bash
python3 skills/contrarian-setup-gate/scripts/run_contrarian_setup_gate.py \
--symbol B6 \
--detector-json reports/cot_crowding_2026-07-12.json \
--news-json reports/nrf_B6_2026-07-12.json \
--price-action-json reports/ta_confirmation_B6_2026-07-12.json \
--as-of 2026-07-15 \
--output-dir reports/
```
`--symbol` and `--detector-json` are required; `--news-json` and `--price-action-json` are optional -- omit either to see the state at that pipeline stage. `--as-of` is required (no implicit "today"): staleness is always evaluated against an explicit reference date so reruns are deterministic.
Exit behavior is intentionally asymmetric: a missing or malformed `--as-of` (or any other CLI usage error) is an operator config mistake, so the CLI exits `2` with usage text and writes no report. A problem with one of the three untrusted report files -- unreadable, malformed, stale, inconsistent -- is always handled fail-closed instead: the CLI exits `0` and writes a report naming the reason, exactly like every other skill in this pipeline.
### Step 2: Read the Setup Status
| Status | Meaning | Next Step |
|---|---|---|
| `READY_FOR_PLAN` | All three steps confirmed; direction, entry_trigger, and invalidation_level are populated | Hand `direction` and `invalidation_level` to a position-sizing skill |
| `WATCHING_PRICE` | Crowding + news confirmed; price action still pending | Run technical-analyst's contrarian-confirmation mode |
| `CROWDED` | Crowding confirmed; news and/or price still pending | Run news-reaction-failure-analyzer |
| `REJECTED` | Crowding is NOT_CONFIRMED (classification NEUTRAL), or news/price came back NOT_CONFIRMED | Stop -- do not run further steps for this symbol/direction |
| `INSUFFICIENT_EVIDENCE` | A required input is missing, unreadable, stale, inconsistent, or could not itself reach a verdict | Stop -- fix or regenerate the named input before rerunning |
`missing_confirmations` lists every step still blocking, with its `state` and `reason`. `warnings` never change the status -- they flag audit-worthy conditions such as a MEDIUM-confidence confirming signal or a near-stale input.
### Step 3: Act on READY_FOR_PLAN Only
At `READY_FOR_PLAN`, `direction` (SHORT/LONG, the fade side of the crowd), `entry_trigger` (a factual echo of the confirming weekly signal), and `invalidation_level` (the stop reference from the price-action report) are populated. `gate_confidence` is the weaker of the news and price-action confidences (HIGH/MEDIUM/LOW -- `LOW` is a token both upstream skills document as reserved but never actually emit; the gate accepts it and ranks it weakest rather than rejecting it as unknown). Position sizing is the next pipeline stage (not yet built as of this skill's release -- see the roadmap in the repository's workflow docs); this gate never places or recommends an order.
## Precedence (Summary)
Each step is evaluated in strict pipeline order -- crowding, then news, then price-action -- and each step fully settles before the next step's file is even consulted. An earlier step's definitive verdict is never softened by a later step's problem.
1. Crowding is evaluated first and exclusively: INVALID/INSUFFICIENT crowding is always `INSUFFICIENT_EVIDENCE`; a NOT_CONFIRMED (NEUTRAL) classification is always `REJECTED`, regardless of any downstream file's state or corruption.
2. With crowding CONFIRMED, news is evaluated next, on its own: INVALID (unreadable, malformed, stale, symbol mismatch, direction mismatch, unsupported schema) forces `INSUFFICIENT_EVIDENCE`; NOT_CONFIRMED forces `REJECTED`; INSUFFICIENT forces `INSUFFICIENT_EVIDENCE` -- in every one of these cases, price-action is never even inspected for the decision.
3. Once news is CONFIRMED (or PENDING, for out-of-order use), price-action is evaluated last, with the same four-way settlement. Running price-action before news (out-of-order pipeline use) caps the status at `CROWDED` with a warning -- a NOT_CONFIRMED price-action verdict still REJECTs even out of order, since price-action is still fully evaluated in that branch.
4. Crowding confirmed, both downstream steps pending -> `CROWDED`.
5. Crowding + news confirmed, price-action pending -> `WATCHING_PRICE`.
6. All three confirmed -> `READY_FOR_PLAN`.
See `references/gate-decision-table.md` for the full decision table (every reachable {crowding} x {news} x {price-action} state combination), the reason-token glossary, and worked examples.
## Output Contract
The script writes `contrarian_setup_gate_<SYMBOL>_<as-of>.json` and `.md` to `--output-dir`:
```yaml
symbol: B6
setup_status: READY_FOR_PLAN | WATCHING_PRICE | CROWDED | REJECTED | INSUFFICIENT_EVIDENCE
direction: SHORT | LONG | null
gate_confidence: HIGH | MEDIUM | LOW | null
entry_trigger: string | null
invalidation_level: number | null
missing_confirmations: [{step, state, reason}, ...]
warnings: [string, ...]
inputs:
crowding: {state, classification, data_date, age_days, report_path}
news_failure: {state, verdict, confidence, verdict_reason, as_of, age_days, report_path}
price_action: {state, verdict, confidence, verdict_reason, stop_reference, as_of, age_days, report_path}
run_context: {symbol, as_of, max_detector_age_days, max_report_age_days, schema_version, skill}
```
Every input's `state` is one of `CONFIRMED`, `NOT_CONFIRMED`, `INSUFFICIENT`, `PENDING` (report not provided), or `INVALID` (a report was provided but is unusable -- unreadable, malformed, stale, or inconsistent with the other inputs; always carries a named reason).
## Guardrails
1. **Never places or recommends orders.** `READY_FOR_PLAN` is the furthest state this skill reaches. Order entry and position sizing are separate, downstream decisions.
2. **INSUFFICIENT_EVIDENCE and REJECTED never advance.** No warning, confidence, or partial input ever pushes the status past what the precedence rules allow.
3. **Fail closed on every input, always.** An unreadable, malformed, stale, symbol-mismatched, or unknown-enum report is never treated as a pass -- it is named and blocks or downgrades the status. This includes the price-action report's `verdict_reason` (allowlisted against technical-analyst's actual confirming-signal vocabulary, not merely type-checked) and its `stop_reference` (must be a finite, positive number -- never non-finite, zero, negative, or a boolean). Before any of the three report files reaches this validation, the CLI also rejects a file outright (reason `<input>_non_finite`) if it contains a non-finite number (`Infinity`/`-Infinity`/`NaN`, including an ordinary-looking number like `1e309` that overflows on parse) ANYWHERE in it, not just in a field the gate reads -- this is what keeps every report a valid, complete JSON file even under adversarial input.
4. **`READY_FOR_PLAN` always carries a usable plan.** `entry_trigger` is guaranteed non-empty and `invalidation_level` is guaranteed a finite positive number whenever the status is `READY_FOR_PLAN` -- enforced both by input validation and by a defensive invariant check.
5. **Not investment advice.** `entry_trigger` and `invalidation_level` are factual echoes of the upstream price-action report, not recommendations.
## Resources
- `scripts/run_contrarian_setup_gate.py` -- CLI: hardened JSON loading (unreadable / parse_error incl. RecursionError / non_finite via an iterative whole-file scan), report generation
- `scripts/gate_logic.py` -- Pure synthesis core: normalization (incl. malformed-shape detection), consistency checks, the precedence state machine
- `references/gate-decision-table.md` -- Full decision table, reason-token glossary, worked examples (including the real B6 REJECTED case)