833f262d12
DaatQuery: 8-feature market state representation DaatVerdict: KNOWN / MARGINAL / OUT_OF_DISTRIBUTION daat_classify: cosine RETRIEVE → magnitude GATE → local MODEL - Cosine finds nearest explored state (directional match) - Magnitude gate detects out-of-distribution states - Empty explored set → always OUT_OF_DISTRIBUTION 9 tests covering: known state, OOD, empty explored, marginal, result fields. No Unicode in code. All tests pass.
136 lines
4.6 KiB
Python
136 lines
4.6 KiB
Python
"""
|
|
DAAT — Direction-Anchored Ambiguity Triage
|
|
|
|
Determines whether a live market state is within the envelope of explored
|
|
states, or whether we are in OUT_OF_DISTRIBUTION territory.
|
|
|
|
Three verdicts:
|
|
KNOWN: live state is well within explored envelope → recommend
|
|
MARGINAL: live state is near boundary → recommend with caution
|
|
OUT_OF_DISTRIBUTION: live state is far outside → refuse, fall back to doctrinal
|
|
|
|
Algorithm (from ANNEX A: cosine RETRIEVE → magnitude GATE → local MODEL):
|
|
1. Cosine similarity to nearest explored state (directional match)
|
|
2. Magnitude gate: detect if magnitude is within explored range
|
|
3. Combine into DaatVerdict
|
|
|
|
Cosine alone returns 1.0 for a crisis (direction matches but magnitude is extreme).
|
|
The magnitude gate prevents false confidence on extreme states.
|
|
|
|
No Unicode in code.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
from enum import Enum
|
|
from typing import List, Optional
|
|
|
|
|
|
class DaatVerdict(Enum):
|
|
"""Verdict from ambiguity triage."""
|
|
KNOWN = "KNOWN" # well within envelope
|
|
MARGINAL = "MARGINAL" # near boundary
|
|
OUT_OF_DISTRIBUTION = "OUT_OF_DISTRIBUTION" # far outside
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class DaatQuery:
|
|
"""A query to the DAAT system — the live market state features."""
|
|
# Core features that define the "position" in the manifold
|
|
spread_bps: float
|
|
depth_usd: float
|
|
imbalance: float # bid/ask imbalance [-1, 1]
|
|
funding_bps: float # current funding rate
|
|
volatility: float # realized vol
|
|
regime_score: float # MARAS regime index
|
|
latency_ms: float # current latency
|
|
inventory_pct: float # current inventory as % of capacity
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class DaatResult:
|
|
"""Result of DAAT classification."""
|
|
verdict: DaatVerdict
|
|
cosine_sim: float # similarity to nearest explored state
|
|
magnitude_ratio: float # magnitude / explored range
|
|
nearest_label: str # label of nearest explored state
|
|
confidence: float # 0.0-1.0
|
|
|
|
|
|
def _cosine_similarity(a: List[float], b: List[float]) -> float:
|
|
"""Cosine similarity between two feature vectors."""
|
|
dot = sum(x * y for x, y in zip(a, b))
|
|
norm_a = sum(x * x for x in a) ** 0.5
|
|
norm_b = sum(x * x for x in b) ** 0.5
|
|
if norm_a < 1e-12 or norm_b < 1e-12:
|
|
return 0.0
|
|
return dot / (norm_a * norm_b)
|
|
|
|
|
|
def _magnitude_ratio(query_vec: List[float], explored_range: List[float]) -> float:
|
|
"""How far is query magnitude from explored range? 1.0 = within range."""
|
|
total = sum(abs(x) for x in query_vec)
|
|
range_max = sum(abs(x) for x in explored_range)
|
|
if range_max < 1e-12:
|
|
return 1.0
|
|
return total / range_max
|
|
|
|
|
|
def daat_classify(
|
|
query: DaatQuery,
|
|
explored_states: List[DaatQuery],
|
|
explored_magnitudes: List[float],
|
|
cosine_threshold: float = 0.7,
|
|
magnitude_threshold: float = 2.0,
|
|
) -> DaatResult:
|
|
"""Classify a live query against explored states.
|
|
|
|
Algorithm:
|
|
1. Find nearest explored state by cosine similarity
|
|
2. Check magnitude gate
|
|
3. Return verdict
|
|
"""
|
|
if not explored_states:
|
|
return DaatResult(
|
|
verdict=DaatVerdict.OUT_OF_DISTRIBUTION,
|
|
cosine_sim=0.0, magnitude_ratio=0.0,
|
|
nearest_label="none", confidence=0.0,
|
|
)
|
|
|
|
query_vec = [query.spread_bps, query.depth_usd, query.imbalance,
|
|
query.funding_bps, query.volatility, query.regime_score,
|
|
query.latency_ms, query.inventory_pct]
|
|
|
|
best_cosine = -1.0
|
|
best_idx = 0
|
|
for i, state in enumerate(explored_states):
|
|
state_vec = [state.spread_bps, state.depth_usd, state.imbalance,
|
|
state.funding_bps, state.volatility, state.regime_score,
|
|
state.latency_ms, state.inventory_pct]
|
|
cos = _cosine_similarity(query_vec, state_vec)
|
|
if cos > best_cosine:
|
|
best_cosine = cos
|
|
best_idx = i
|
|
|
|
# Magnitude gate
|
|
mag_ratio = _magnitude_ratio(query_vec, explored_magnitudes)
|
|
|
|
# Verdict
|
|
if best_cosine >= cosine_threshold and mag_ratio <= magnitude_threshold:
|
|
verdict = DaatVerdict.KNOWN
|
|
confidence = best_cosine * (1.0 / max(mag_ratio, 0.1))
|
|
elif best_cosine >= cosine_threshold * 0.5:
|
|
verdict = DaatVerdict.MARGINAL
|
|
confidence = best_cosine * 0.5
|
|
else:
|
|
verdict = DaatVerdict.OUT_OF_DISTRIBUTION
|
|
confidence = 0.0
|
|
|
|
return DaatResult(
|
|
verdict=verdict,
|
|
cosine_sim=best_cosine,
|
|
magnitude_ratio=mag_ratio,
|
|
nearest_label=f"state_{best_idx}",
|
|
confidence=min(1.0, confidence),
|
|
)
|