"""Pure candidate -> wire-signal mapping (v2: lenient + loud). candidate_to_signal( candidate, produced_at_ms, *, signal_id=None, # stable production id (journal eid); minted if absent signal_data_schema, # dict {key: {type, required?}} | None default_signal_validity_seconds, # int REQUIRED fallback TTL (seconds) on_reject=None, # callback(reason_dict) on reject -> /errors ) -> dict | None Returns the wire dict on success: {signal_id, asset, direction, produced_at, valid_until, data} plus signal_type only when the author set a non-empty string, and the canonical sizing fields `marginPct` (percent 0–100 of withdrawable, not a fraction) and/or `leverage` at the TOP LEVEL (not inside `data`) when the candidate sets them — the runtime reads signal.marginPct / signal.leverage directly. A present-but-non-positive leverage, or a marginPct outside (0, 100], is a LOUD reject. Returns None (never raises for a bad CANDIDATE) when the candidate is invalid, and — when on_reject is supplied — calls on_reject(reason) so the rejection is LOUD (logged AND posted to /errors), never a silent drop. v2 contract (design §4): - direction is NORMALIZED case: "long"/"LONG"/"Long" -> "LONG"; "short"/.. -> "SHORT". Surrounding whitespace is stripped. "buy"/"sell"/""/non-string -> REJECT. - asset + direction are MANDATORY (non-empty string asset). - `data` is VALIDATED against signal_data_schema (when a dict is supplied): missing-required / wrong-type / extra (unknown) keys all REJECT. A None schema means "no validation" (passthrough — used when no schema is threaded in). - valid_until = produced_at_ms + (valid_for_seconds else default_signal_validity_seconds) * 1000. The v1 produced_at + 2*interval default is GONE; there is no interval_ms arg. A non-positive / non-int valid_for_seconds is REJECTED. A MISSING default_signal_validity_seconds (None) when it is NEEDED fails LOUD (raises) — no magic fallback. - author absolute `valid_until` is DROPPED (never honored, never leaks). - The keys on the wire are {signal_id, asset, direction, produced_at, valid_until, data} (+ optional signal_type, marginPct, leverage). `marginPct`/`leverage` are the canonical top-level sizing fields (validated when present; `marginPct` is a percent 0–100 of withdrawable — not a fraction, not a USD value). `signal_id` is the scaffold-minted stable production id (the journal eid), threaded in by the delivery sink — it is the intake dedup key. An author-supplied `signal_id` in the candidate dict is IGNORED: the scaffold owns the dedup key, an author cannot spoof it. Apart from signal_id/signal_type, author bookkeeping never leaks. """ from __future__ import annotations import uuid from typing import Any, Optional # JSON type name -> Python type(s) for schema validation. `bool` is excluded from # "number" because a bool is an int subclass in Python but is not a numeric value. def _is_number(value: Any) -> bool: return isinstance(value, (int, float)) and not isinstance(value, bool) def _is_string(value: Any) -> bool: return isinstance(value, str) def _is_boolean(value: Any) -> bool: return isinstance(value, bool) def _is_object(value: Any) -> bool: return isinstance(value, dict) def _is_array(value: Any) -> bool: return isinstance(value, list) _TYPE_CHECKS = { "number": _is_number, "string": _is_string, "boolean": _is_boolean, "object": _is_object, "array": _is_array, } def _validate_data_against_schema(data: dict, schema: dict) -> Optional[str]: """Validate `data` against `schema`. Returns a reason string on violation, or None when valid. additionalProperties:false — unknown keys reject.""" allowed = set(schema.keys()) # Extra (unknown) keys reject. for key in data: if key not in allowed: return f"data has unknown key {key!r} (not in signal_data_schema)" # Missing-required and wrong-type. for key, spec in schema.items(): spec = spec if isinstance(spec, dict) else {} required = spec.get("required", True) # required unless explicitly False if key not in data: if required: return f"data missing required key {key!r}" continue declared_type = spec.get("type") check = _TYPE_CHECKS.get(declared_type) if check is not None and not check(data[key]): return ( f"data key {key!r} has wrong type " f"(expected {declared_type}, got {type(data[key]).__name__})" ) return None # Top-level candidate keys the envelope recognizes and lifts onto the wire signal. # Anything else an author sets at the top level is dropped (it never leaks to the # wire by design) — `on_warn` names it so a removed/typo'd sizing field (e.g. # `marginUsd`, or `marginpct` miscased) is visible instead of vanishing silently. _RECOGNIZED_CANDIDATE_KEYS = frozenset( { "asset", "direction", "valid_for_seconds", "data", "leverage", "marginPct", "signal_type", "signal_id", # author-supplied id is intentionally ignored; don't warn on it } ) def candidate_to_signal( candidate: Any, produced_at_ms: int, *, signal_id: Optional[str] = None, signal_data_schema: Optional[dict], default_signal_validity_seconds: Optional[int], on_reject=None, on_warn=None, ) -> dict | None: """Wrap a scan() candidate into a wire signal dict (v2). Returns the signal dict on success, None when the candidate is invalid (and calls on_reject(reason) so the rejection is loud). Raises only when the OPERATOR config is wrong (missing default_signal_validity_seconds when needed). """ def _reject(reason: str) -> None: if on_reject is not None: on_reject({"reason": reason, "candidate": repr(candidate)}) if not isinstance(candidate, dict): _reject(f"candidate is not a mapping (got {type(candidate).__name__})") return None # -- asset: mandatory, non-empty string -------------------------------- asset = candidate.get("asset") if not isinstance(asset, str) or not asset.strip(): _reject(f"asset missing/blank/non-string (got {asset!r})") return None # -- direction: mandatory string, case-normalized to LONG / SHORT ------ direction = candidate.get("direction") if not isinstance(direction, str) or isinstance(direction, bool): _reject(f"direction is not a string (got {direction!r})") return None normalized = direction.strip().upper() if normalized not in ("LONG", "SHORT"): _reject(f"direction must be long/short (got {direction!r})") return None # -- valid_until from valid_for_seconds else the default --------------- # (compute BEFORE schema so a missing operator default still fails loud.) if "valid_for_seconds" in candidate: vfs = candidate["valid_for_seconds"] # int only (bool excluded), strictly positive. if not isinstance(vfs, int) or isinstance(vfs, bool) or vfs <= 0: _reject(f"valid_for_seconds must be a positive int (got {vfs!r})") return None validity_seconds = vfs else: if default_signal_validity_seconds is None: # OPERATOR error, not a candidate bug — fail LOUD (no magic fallback). raise ValueError( "default_signal_validity_seconds is required when a candidate has " "no valid_for_seconds; got None (no magic fallback in v2)." ) validity_seconds = default_signal_validity_seconds valid_until = produced_at_ms + validity_seconds * 1000 # -- data: validate against schema (when supplied) --------------------- data = candidate.get("data", {}) if not isinstance(data, dict): data = {} if signal_data_schema is not None: violation = _validate_data_against_schema(data, signal_data_schema) if violation is not None: _reject(violation) return None # -- canonical sizing: marginPct / leverage ride at the TOP LEVEL of the wire # signal, NOT inside `data`. `marginPct` is a percent (0–100] of withdrawable # (NOT a fraction, NOT a USD value); the runtime consumes signal.marginPct / # signal.leverage directly (data is freeform context only), so per-signal # sizing/leverage from the producer is honored instead of silently dropped. # Present-but-invalid is a LOUD reject — sizing must never fall back silently. sizing: dict = {} if candidate.get("leverage") is not None: value = candidate["leverage"] if not _is_number(value) or value <= 0: _reject(f"leverage must be a positive number when set (got {value!r})") return None sizing["leverage"] = value # marginPct is a PERCENT of withdrawable (0–100], not a USD value and not a # fraction — same units/basis as the runtime's config `margin_pct`. Bounded # (fail-closed) so an out-of-range percent never sizes silently. margin_pct = candidate.get("marginPct") if margin_pct is not None: if not _is_number(margin_pct) or margin_pct <= 0 or margin_pct > 100: _reject( f"marginPct must be a number in (0, 100] when set (got {margin_pct!r})" ) return None sizing["marginPct"] = margin_pct sig: dict = { # scaffold-minted stable production id (journal eid); the author cannot # override it via the candidate dict — the kwarg is authoritative. The # delivery sink threads the eid in; when no id is supplied a stable one is # minted so the wire ALWAYS carries a signal_id (required end-to-end). "signal_id": signal_id if (isinstance(signal_id, str) and signal_id) else uuid.uuid4().hex, "asset": asset, "direction": normalized, "produced_at": produced_at_ms, "valid_until": valid_until, "data": data, } # Lift the validated sizing fields onto the wire signal at the top level. sig.update(sizing) signal_type = candidate.get("signal_type") if signal_type and isinstance(signal_type, str): sig["signal_type"] = signal_type # The signal IS valid and will emit — but warn (don't reject) about any # top-level key we did not lift, so a producer whose sizing field was dropped # (removed/typo'd) sees why instead of debugging a silently mis-sized signal. if on_warn is not None: dropped = sorted(k for k in candidate if k not in _RECOGNIZED_CANDIDATE_KEYS) if dropped: on_warn({"dropped": dropped, "asset": asset}) return sig