SAS · PHASE 2 · DATA ACQUISITION & MARKET STATE FOUNDATION ROUND 1 · FOUNDATION

SAS TRADING SYSTEM · ROUND 1 · PHASE 2

Data Acquisition &
Market State Foundation
Complete Implementation Guide · All 7 Deliverables · Full Production Code
Binance REST · WebSocket Manager · Order Book · Derivatives · Sentiment · Data Quality · Replay Engine

Del Title Output

ive
rab
le

D1 Binance REST Connector 2-year OHLCV backfill · Checkpoint/resume · Upsert to TimescaleDB

D2 Binance WebSocket Manager Persistent live streams · Auto-reconnect · Backpressure queue

D3 Order Book Snapshot System 1-second L2 snapshots · Imbalance ratio · Redis + TimescaleDB

D4 Derivatives Data Pipeline Funding rates · Open Interest · Liquidations · Long/Short ratios

D5 Sentiment & On-Chain Pipeline Fear & Greed · Glassnode · CryptoCompare NLP · Daily ingestion

D6 Data Quality Monitor & Auto- Gap detection · Auto-backfill · data_quality_events · DLQ alerts
Backfill

D7 Replay Engine Zero look-ahead MarketState iterator · Sole backtest data source

The Most Critical Phase in Round 1

Every signal, backtest result, and ML model in all 24 downstream phases depends entirely on what is built here.
Bad data in Phase 2 means wrong signals in Phase 4, corrupted backtests in Phase 7, and poisoned ML models
in Phase 21. There is no recovery from a broken data foundation. Phase 2 demands disproportionate
investment in correctness.

Phase Start: ________________ · Target End: ________________ · Actual Hours: ______

SAS Phase 2 Implementation Guide · 2026 All 7 Deliverables · Full Production Code
SAS · PHASE 2 · DATA ACQUISITION & MARKET STATE FOUNDATION ROUND 1 · FOUNDATION

Data Acquisition & Market State Foundation

P2 Round 1 · 3–5 Weeks · ~175–250 Hours · Role: Data Engineer
Depends On: Phase 1 complete · Output: MarketState in Redis

P2 / 25 R1 — 3–5 Weeks ~175–250 7 (D1–D7) MarketState →

Phase Foundation Duration Hours Deliverables Redis
Round Key Output

Phase 2 Overview
Phase 2 builds the data nervous system. It establishes every external data connection, persists all historical
and live market data to TimescaleDB, publishes real-time MarketState to Redis, and provides the Replay
Engine that all downstream backtesting will use. The 15 external data sources, compression policies, quality
monitoring, and auto-backfill logic are all built here.

C5: C5 Canonical Correction — Applied Here

TimescaleDB compression policies are configured in Phase 2 — not deferred to Phase 17 as the original
directive stated. Every hypertable gets its compression policy at creation time. Deferring this to Phase 17 would
cause query performance degradation across Rounds 1 through 3 as tables grow.

Prerequisites Checklist
C Prerequisite Verification
h
e
c
k

☐ Phase 1 complete — all 8 deliverables docker compose ps → all 3 healthy. curl localhost:8000/health →
verified {status:'healthy'}

☐ Binance Testnet account active with API [Link] — login works. Testnet key + secret in .env
keys generated

☐ TimescaleDB tables exist (from Phase 1 psql → \dt → shows ohlcv, orderbook_snapshots, funding_rates,
migration) open_interest

☐ Redis accessible and responding to PING docker exec sas-cache redis-cli ping → PONG

☐ python-binance or ccxt library chosen and pip install python-binance OR pip install ccxt — add to [Link]
installed

☐ Coinglass, Glassnode, CryptoCompare API Keys from D4 of Phase 0 stored in .env (not committed to git)
keys in .env

☐ [Link] URL accessible from VPS curl [Link] → returns JSON with value field

The MarketState Object — What Phase 2 Produces

SAS Phase 2 Implementation Guide · 2026 All 7 Deliverables · Full Production Code
SAS · PHASE 2 · DATA ACQUISITION & MARKET STATE FOUNDATION ROUND 1 · FOUNDATION

Every downstream phase consumes MarketState objects. This is the central data structure of the entire system.
Phase 2 is responsible for building it correctly and publishing it to Redis on every candle close and every order
book update.

sas/data/[Link]
# sas/data/[Link]
from __future__ import annotations
from dataclasses import dataclass, field
from datetime import datetime, timezone
from decimal import Decimal
from typing import Optional

@dataclass
class OHLCVBar:
time: datetime
symbol: str
timeframe: str
open: Decimal
high: Decimal
low: Decimal
close: Decimal
volume: Decimal

@dataclass
class OrderBookState:
time: datetime
symbol: str
bid_volume_top5: Decimal
ask_volume_top5: Decimal
imbalance_ratio: float # bid_vol / ask_vol
best_bid: Decimal
best_ask: Decimal
spread_pct: float
is_stale: bool = False

@dataclass
class DerivativesState:
time: datetime
symbol: str
funding_rate: float
funding_z_score: float # vs 30-day rolling mean
open_interest: Decimal
oi_delta_4h: float # OI change over last 4 hours
long_short_ratio: float
is_stale: bool = False

@dataclass
class SentimentState:
time: datetime
fear_greed_index: int # 0–100

SAS Phase 2 Implementation Guide · 2026 All 7 Deliverables · Full Production Code
SAS · PHASE 2 · DATA ACQUISITION & MARKET STATE FOUNDATION ROUND 1 · FOUNDATION

fear_greed_z: float # vs 90-day rolling mean

exchange_net_flow: float # positive = inflow (bearish), negative = outflow
news_sentiment: float # 0.0–1.0, 24h rolling NLP
is_stale: bool = False

@dataclass
class MarketState:
"""Complete market snapshot at a specific timestamp.

Published to Redis on every 1M candle close (scalp engine reads)

and every 4H candle close (swing engine reads).

For backtesting: only includes data available at [Link].

Zero look-ahead guarantee is enforced by the Replay Engine.
"""
timestamp: datetime
symbol: str

# OHLCV per timeframe — keyed by '1m', '5m', '15m', '1h', '4h', '1d'
candles: dict[str, list[OHLCVBar]] = field(default_factory=dict)

# Current order book (None if stale or unavailable)

order_book: Optional[OrderBookState] = None

# Derivatives data
derivatives: Optional[DerivativesState] = None

# Sentiment data
sentiment: Optional[SentimentState] = None

# ATR values per timeframe (pre-computed for Phase 5 TA)

atr: dict[str, float] = field(default_factory=dict)

# Data quality flags

has_gaps: bool = False
gap_details: list[str] = field(default_factory=list)

SAS Phase 2 Implementation Guide · 2026 All 7 Deliverables · Full Production Code
SAS · PHASE 2 · DATA ACQUISITION & MARKET STATE FOUNDATION ROUND 1 · FOUNDATION

Binance REST Connector

D1 Full OHLCV backfill · Checkpoint/resume · Upsert to TimescaleDB

The REST connector performs the 2-year historical backfill for all timeframes (1m, 5m, 15m, 1h, 4h, 1d) for both
BTCUSDT and ETHUSDT. It uses an upsert pattern to prevent duplicate rows on restart, and a
checkpoint/resume mechanism so an interrupted backfill picks up exactly where it left off.

Backfill Time Estimate

A full 2-year 1M candle backfill for 2 symbols is approximately 2 million rows. At Binance's 1200 request/minute
weight limit, this takes 3–6 hours. Do not interrupt it — the checkpoint system ensures it resumes correctly. Run
the backfill overnight on first deployment.

sas/data/binance_rest.py
sas/data/binance_rest.py
from __future__ import annotations

import asyncio
import json
from datetime import datetime, timezone, timedelta
from decimal import Decimal
from pathlib import Path
from typing import AsyncIterator

import asyncpg
from binance import AsyncClient

from [Link] import settings

from [Link] import get_logger
from [Link] import bus, DataGapDetectedEvent

logger = get_logger(__name__)

SYMBOLS = ['BTCUSDT', 'ETHUSDT']

TIMEFRAMES = ['1m', '5m', '15m', '1h', '4h', '1d']
BACKFILL_YEARS = 2
CHECKPOINT_FILE = Path('/app/data/.backfill_checkpoint.json')

# Binance kline interval map

TF_MAP = {'1m':'1m','5m':'5m','15m':'15m','1h':'1h','4h':'4h','1d':'1d'}

class BinanceRestConnector:
"""Handles all REST-based data operations.

Responsibilities:
- Full historical OHLCV backfill with checkpoint/resume
- Periodic funding rate, OI, and liquidation fetches
- Gap-fill for missing candles detected by Data Quality Monitor
- All writes use UPSERT (ON CONFLICT DO NOTHING) to prevent duplicates
"""

SAS Phase 2 Implementation Guide · 2026 All 7 Deliverables · Full Production Code
SAS · PHASE 2 · DATA ACQUISITION & MARKET STATE FOUNDATION ROUND 1 · FOUNDATION

def __init__(self, pool: [Link]) -> None:

self._pool = pool
self._client: AsyncClient | None = None

async def start(self) -> None:

self._client = await [Link](
api_key=settings.binance_testnet_api_key if settings.binance_testnet
else settings.binance_api_key,
api_secret=settings.binance_testnet_secret_key if settings.binance_testnet
else settings.binance_secret_key,
testnet=settings.binance_testnet,
)
[Link]('binance_rest_started', testnet=settings.binance_testnet)

async def stop(self) -> None:

if self._client:
await self._client.close_connection()

# ── BACKFILL ──────────────────────────────────────────────────
async def run_full_backfill(self) -> None:
"""Run 2-year OHLCV backfill for all symbols + timeframes.

Reads checkpoint file to skip already-completed symbol/timeframe

combinations. Safe to interrupt and restart at any time.
"""
checkpoint = self._load_checkpoint()
end_time = [Link]([Link])
start_time = end_time - timedelta(days=365 * BACKFILL_YEARS)

for symbol in SYMBOLS:

for tf in TIMEFRAMES:
key = f'{symbol}:{tf}'
if [Link](key) == 'complete':
[Link]('backfill_skipped', symbol=symbol, timeframe=tf)
continue

resume_from = [Link](key)
from_dt = [Link](resume_from) if resume_from \
else start_time

[Link]('backfill_starting', symbol=symbol, timeframe=tf,

from_dt=str(from_dt))
await self._backfill_symbol_tf(symbol, tf, from_dt, end_time)
checkpoint[key] = 'complete'
self._save_checkpoint(checkpoint)
[Link]('backfill_complete', symbol=symbol, timeframe=tf)

async def _backfill_symbol_tf(

self, symbol: str, tf: str,
start: datetime, end: datetime,
) -> None:
current = start
batch_size = 1000 # Binance max per request

SAS Phase 2 Implementation Guide · 2026 All 7 Deliverables · Full Production Code
SAS · PHASE 2 · DATA ACQUISITION & MARKET STATE FOUNDATION ROUND 1 · FOUNDATION

while current < end:

try:
klines = await self._client.get_historical_klines(
symbol=symbol,
interval=TF_MAP[tf],
start_str=int([Link]() * 1000),
limit=batch_size,
)
if not klines:
break

await self._upsert_klines(symbol, tf, klines)

last_ts = klines[-1][0] # open time ms

current = [Link](last_ts/1000, tz=[Link])
current += timedelta(seconds=1)

# Rate limit — respect Binance weight limits

await [Link](0.1)

except Exception as exc:

[Link]('backfill_error', symbol=symbol, tf=tf,
current=str(current), error=str(exc))
await [Link](5) # Back off on error
raise

async def _upsert_klines(

self, symbol: str, tf: str, klines: list
) -> None:
rows = [(
[Link](k[0]/1000, tz=[Link]), # time
symbol, # symbol
tf, # timeframe
Decimal(k[1]), # open
Decimal(k[2]), # high
Decimal(k[3]), # low
Decimal(k[4]), # close
Decimal(k[5]), # volume
) for k in klines]

async with self._pool.acquire() as conn:

await [Link](
'''INSERT INTO ohlcv
(time, symbol, timeframe, open, high, low, close, volume)
VALUES ($1,$2,$3,$4,$5,$6,$7,$8)
ON CONFLICT (time, symbol, timeframe) DO NOTHING''',
rows,
)

# ── CHECKPOINT ────────────────────────────────────────────────
def _load_checkpoint(self) -> dict:
if CHECKPOINT_FILE.exists():
return [Link](CHECKPOINT_FILE.read_text())
return {}

SAS Phase 2 Implementation Guide · 2026 All 7 Deliverables · Full Production Code
SAS · PHASE 2 · DATA ACQUISITION & MARKET STATE FOUNDATION ROUND 1 · FOUNDATION

def _save_checkpoint(self, data: dict) -> None:

CHECKPOINT_FILE.[Link](parents=True, exist_ok=True)
CHECKPOINT_FILE.write_text([Link](data, indent=2))

# ── GAP FILL ──────────────────────────────────────────────────

async def fill_gap(
self, symbol: str, tf: str,
gap_start: datetime, gap_end: datetime,
) -> int:
"""Fill a specific OHLCV gap detected by the quality monitor.

Called automatically by DataQualityMonitor when a gap is detected.

Returns number of rows inserted.
"""
[Link]('gap_fill_starting', symbol=symbol, tf=tf,
gap_start=str(gap_start), gap_end=str(gap_end))
klines = await self._client.get_historical_klines( # type: ignore[union-attr]
symbol=symbol,
interval=TF_MAP[tf],
start_str=int(gap_start.timestamp() * 1000),
end_str=int(gap_end.timestamp() * 1000),
limit=1000,
)
if klines:
await self._upsert_klines(symbol, tf, klines)
[Link]('gap_fill_complete', rows=len(klines))
return len(klines)

SAS Phase 2 Implementation Guide · 2026 All 7 Deliverables · Full Production Code
SAS · PHASE 2 · DATA ACQUISITION & MARKET STATE FOUNDATION ROUND 1 · FOUNDATION

Binance WebSocket Manager

D2 Persistent live streams · Auto-reconnect · Backpressure queue

The WebSocket manager maintains persistent connections for live kline data and L2 order book snapshots. It
implements exponential backoff reconnection, a watchdog task that detects stale connections, and an
[Link] with backpressure to buffer bursts without memory exhaustion.

sas/data/websocket_manager.py
sas/data/websocket_manager.py
from __future__ import annotations

import asyncio
import json
import time
from datetime import datetime, timezone
from decimal import Decimal
from typing import Callable, Awaitable

import websockets
from [Link] import ConnectionClosed

from [Link] import settings

from [Link] import get_logger
from [Link] import bus, DataGapDetectedEvent

logger = get_logger(__name__)

SYMBOLS = ['btcusdt', 'ethusdt']

TIMEFRAMES = ['1m', '5m', '15m', '1h', '4h', '1d']

# Binance stream URLs

WS_BASE = '[Link]
WS_TESTNET = '[Link]

# Backpressure: reject new messages if queue exceeds this size

QUEUE_MAXSIZE = 10_000

# Watchdog: reconnect if no message received in this many seconds

STALE_THRESHOLD_SEC = 10

KlineHandler = Callable[[dict], Awaitable[None]]

OrderBookHandler= Callable[[dict], Awaitable[None]]

class WebSocketManager:
"""Manages persistent Binance WebSocket connections.

Maintains streams for:

- Klines (all timeframes, all symbols)
- L2 order book depth (20 levels, all symbols)

SAS Phase 2 Implementation Guide · 2026 All 7 Deliverables · Full Production Code
SAS · PHASE 2 · DATA ACQUISITION & MARKET STATE FOUNDATION ROUND 1 · FOUNDATION

Implements:
- Exponential backoff reconnection (1s → 2s → 4s → 8s → 16s cap)
- Watchdog task that kills and restarts stale connections
- [Link] with maxsize=10,000 for backpressure
"""

def __init__(
self,
on_kline: KlineHandler,
on_orderbook: OrderBookHandler,
) -> None:
self._on_kline = on_kline
self._on_orderbook = on_orderbook
self._queue: [Link][dict] = [Link](maxsize=QUEUE_MAXSIZE)
self._running = False
self._last_message_ts: dict[str, float] = {} # stream_name -> timestamp
self._tasks: list[[Link]] = []

async def start(self) -> None:

self._running = True
ws_base = WS_TESTNET if settings.binance_testnet else WS_BASE

# Build combined stream URL for all klines + order books

streams = []
for sym in SYMBOLS:
for tf in TIMEFRAMES:
[Link](f'{sym}@kline_{tf}')
[Link](f'{sym}@depth20@100ms') # 100ms order book updates

stream_path = '/'.join(streams)
url = f'{ws_base}?streams={stream_path}'

# Launch connection task and watchdog

self._tasks.append(asyncio.create_task(
self._connection_loop(url), name='ws_connection'
))
self._tasks.append(asyncio.create_task(
self._watchdog_loop(), name='ws_watchdog'
))
self._tasks.append(asyncio.create_task(
self._process_queue_loop(), name='ws_queue_processor'
))
[Link]('websocket_manager_started', streams=len(streams))

async def stop(self) -> None:

self._running = False
for task in self._tasks:
[Link]()
await [Link](*self._tasks, return_exceptions=True)
[Link]('websocket_manager_stopped')

# ── CONNECTION LOOP ───────────────────────────────────────────

async def _connection_loop(self, url: str) -> None:
backoff = 1

SAS Phase 2 Implementation Guide · 2026 All 7 Deliverables · Full Production Code
SAS · PHASE 2 · DATA ACQUISITION & MARKET STATE FOUNDATION ROUND 1 · FOUNDATION

while self._running:
try:
[Link]('websocket_connecting', url=url[:60])
async with [Link](
url,
ping_interval=20,
ping_timeout=10,
close_timeout=5,
) as ws:
backoff = 1 # Reset on successful connection
[Link]('websocket_connected')
async for raw in ws:
if not self._running:
break
try:
msg = [Link](raw)
self._last_message_ts[[Link]('stream','')]= [Link]()
# Non-blocking put — drop if queue full (backpressure)
try:
self._queue.put_nowait(msg)
except [Link]:
[Link]('ws_queue_full_dropping_message')
except [Link]:
pass

except (ConnectionClosed, OSError, [Link]) as exc:

[Link]('websocket_disconnected', error=str(exc),
reconnect_in=backoff)
await [Link](backoff)
backoff = min(backoff * 2, 16) # Cap at 16s

except [Link]:
break

# ── WATCHDOG ─────────────────────────────────────────────────
async def _watchdog_loop(self) -> None:
while self._running:
await [Link](15)
now = [Link]()
for stream, last_ts in list(self._last_message_ts.items()):
age = now - last_ts
if age > STALE_THRESHOLD_SEC:
[Link]('websocket_stream_stale',
stream=stream, age_sec=round(age, 1))
# Phase 9 latency check is handled by DataQualityMonitor
await [Link](DataGapDetectedEvent(
source='websocket',
description=f'Stream {stream} stale for {age:.1f}s',
))

# ── QUEUE PROCESSOR ──────────────────────────────────────────

async def _process_queue_loop(self) -> None:
while self._running:
try:
msg = await asyncio.wait_for(self._queue.get(), timeout=1.0)

SAS Phase 2 Implementation Guide · 2026 All 7 Deliverables · Full Production Code
SAS · PHASE 2 · DATA ACQUISITION & MARKET STATE FOUNDATION ROUND 1 · FOUNDATION

stream = [Link]('stream', '')

data = [Link]('data', {})

if '@kline_' in stream:
await self._on_kline(data)
elif '@depth' in stream:
await self._on_orderbook(data)

except [Link]:
continue # Normal — no messages in 1s
except [Link]:
break
except Exception as exc:
[Link]('queue_processor_error', error=str(exc))

SAS Phase 2 Implementation Guide · 2026 All 7 Deliverables · Full Production Code
SAS · PHASE 2 · DATA ACQUISITION & MARKET STATE FOUNDATION ROUND 1 · FOUNDATION

Order Book Snapshot System

D3 1-second L2 snapshots · Imbalance ratio · Redis TTL + TimescaleDB

The order book snapshot system captures L2 depth every time the WebSocket pushes an update
(approximately every 100ms), computes the bid/ask imbalance ratio, stores the snapshot in Redis with a 10-
second TTL for real-time access by the Phase 4 Scalp Engine, and persists snapshots to TimescaleDB for
backtesting.

Imbalance Ratio — The Scalp Engine's Primary Signal Input

The bid/ask volume imbalance ratio is the most time-sensitive input in the entire system. The Scalp Engine
checks it at signal generation time. An OB imbalance ratio > 1.5 (bid-heavy) is a prerequisite for long scalp
entries. This value must be computed from the live order book with zero staleness tolerance.

sas/data/[Link]
sas/data/[Link]
from __future__ import annotations

import json
from datetime import datetime, timezone
from decimal import Decimal

import asyncpg
import [Link] as aioredis

from [Link] import get_logger

from [Link] import OrderBookState

logger = get_logger(__name__)

# Redis key pattern: 'ob:{symbol}' e.g. 'ob:BTCUSDT'

REDIS_KEY_TTL = 10 # seconds — stale after 10s

# How many levels to use for imbalance ratio

DEPTH_LEVELS = 5

class OrderBookProcessor:
"""Processes live order book WebSocket messages.

For each update:

1. Computes bid/ask imbalance ratio from top-5 levels
2. Writes snapshot to Redis with 10-second TTL
3. Persists to TimescaleDB for backtest replay
4. Publishes OrderBookState for consumption by Phase 4
"""

def __init__(self, pool: [Link], redis: [Link]) -> None:

self._pool = pool
self._redis = redis

SAS Phase 2 Implementation Guide · 2026 All 7 Deliverables · Full Production Code
SAS · PHASE 2 · DATA ACQUISITION & MARKET STATE FOUNDATION ROUND 1 · FOUNDATION

async def handle_depth_message(self, data: dict) -> None:

"""Process a @depth20 WebSocket message."""
try:
symbol = [Link]('s', '').upper() # e.g. 'BTCUSDT'
if not symbol:
return

bids = [Link]('b', [])[:DEPTH_LEVELS] # [[price, qty], ...]

asks = [Link]('a', [])[:DEPTH_LEVELS]

if not bids or not asks:

return

bid_vol = sum(Decimal(b[1]) for b in bids)

ask_vol = sum(Decimal(a[1]) for a in asks)

imbalance = float(bid_vol / ask_vol) if ask_vol else 1.0

best_bid = Decimal(bids[0][0])
best_ask = Decimal(asks[0][0])
spread_pct= float((best_ask - best_bid) / best_bid * 100)

snapshot = OrderBookState(
time = [Link]([Link]),
symbol = symbol,
bid_volume_top5 = bid_vol,
ask_volume_top5 = ask_vol,
imbalance_ratio = imbalance,
best_bid = best_bid,
best_ask = best_ask,
spread_pct = spread_pct,
)

# Write to Redis (sub-millisecond reads for Scalp Engine)

await self._write_to_redis(symbol, snapshot)

# Persist to TimescaleDB (async, non-blocking)

await self._persist_snapshot(snapshot)

except Exception as exc:

[Link]('orderbook_process_error', error=str(exc))

async def _write_to_redis(self, symbol: str, ob: OrderBookState) -> None:

key = f'ob:{symbol}'
payload = {
'time': [Link](),
'symbol': [Link],
'bid_volume_top5': str(ob.bid_volume_top5),
'ask_volume_top5': str(ob.ask_volume_top5),
'imbalance_ratio': ob.imbalance_ratio,
'best_bid': str(ob.best_bid),
'best_ask': str(ob.best_ask),
'spread_pct': ob.spread_pct,
}
await self._redis.setex(key, REDIS_KEY_TTL, [Link](payload))

SAS Phase 2 Implementation Guide · 2026 All 7 Deliverables · Full Production Code
SAS · PHASE 2 · DATA ACQUISITION & MARKET STATE FOUNDATION ROUND 1 · FOUNDATION

async def _persist_snapshot(self, ob: OrderBookState) -> None:

async with self._pool.acquire() as conn:
await [Link](
'''INSERT INTO orderbook_snapshots
(time, symbol, bid_volume_top5, ask_volume_top5,
imbalance_ratio, best_bid, best_ask, spread_pct)
VALUES ($1,$2,$3,$4,$5,$6,$7,$8)
ON CONFLICT DO NOTHING''',
[Link], [Link],
ob.bid_volume_top5, ob.ask_volume_top5,
ob.imbalance_ratio,
ob.best_bid, ob.best_ask,
ob.spread_pct,
)

async def get_current(self, symbol: str) -> OrderBookState | None:

"""Read current order book from Redis (used by Phase 4 Scalp Engine)."""
raw = await self._redis.get(f'ob:{symbol}')
if not raw:
return None
data = [Link](raw)
return OrderBookState(
time = [Link](data['time']),
symbol = data['symbol'],
bid_volume_top5 = Decimal(data['bid_volume_top5']),
ask_volume_top5 = Decimal(data['ask_volume_top5']),
imbalance_ratio = data['imbalance_ratio'],
best_bid = Decimal(data['best_bid']),
best_ask = Decimal(data['best_ask']),
spread_pct = data['spread_pct'],
)

SAS Phase 2 Implementation Guide · 2026 All 7 Deliverables · Full Production Code
SAS · PHASE 2 · DATA ACQUISITION & MARKET STATE FOUNDATION ROUND 1 · FOUNDATION

Derivatives Data Pipeline

D4 Funding rates · Open Interest · Liquidations · Long/Short ratios

The derivatives pipeline ingests funding rates (every 8 hours), open interest (every 15 minutes), liquidation
levels, and long/short ratios from Binance and Coinglass. All data is stored as TimescaleDB hypertables with
compression configured here (C5 correction). The Z-score calculation for funding rate normalization is pre-
computed on ingestion.

sas/data/[Link]
sas/data/[Link]
from __future__ import annotations

import asyncio
from datetime import datetime, timezone, timedelta
from decimal import Decimal
from statistics import mean, stdev

import asyncpg
import httpx

from [Link] import settings

from [Link] import get_logger
from [Link] import DerivativesState

logger = get_logger(__name__)

SYMBOLS = ['BTCUSDT', 'ETHUSDT']

COINGLASS_BASE = '[Link]

class DerivativesPipeline:
"""Periodic ingestion of derivatives data.

Schedule:
- Funding rates: every 8 hours (or on each payment time)
- Open Interest: every 15 minutes
- Long/Short ratio: every 15 minutes (Coinglass)
- Liquidations: every 1 hour

C5 Correction: Compression policies are set here (Phase 2),

NOT deferred to Phase 17.
"""

def __init__(self, pool: [Link], http: [Link]) -> None:

self._pool = pool
self._http = http
self._tasks: list[[Link]] = []

async def start(self) -> None:

self._tasks.append(asyncio.create_task(
self._funding_loop(), name='funding_loop'

SAS Phase 2 Implementation Guide · 2026 All 7 Deliverables · Full Production Code
SAS · PHASE 2 · DATA ACQUISITION & MARKET STATE FOUNDATION ROUND 1 · FOUNDATION

))
self._tasks.append(asyncio.create_task(
self._oi_loop(), name='oi_loop'
))
self._tasks.append(asyncio.create_task(
self._ls_ratio_loop(), name='ls_ratio_loop'
))
[Link]('derivatives_pipeline_started')

async def stop(self) -> None:

for t in self._tasks: [Link]()
await [Link](*self._tasks, return_exceptions=True)

# ── FUNDING RATES ─────────────────────────────────────────────

async def _funding_loop(self) -> None:
while True:
try:
for symbol in SYMBOLS:
await self._fetch_and_store_funding(symbol)
except Exception as exc:
[Link]('funding_loop_error', error=str(exc))
await [Link](60 * 60 * 8) # 8 hours

async def _fetch_and_store_funding(self, symbol: str) -> None:

url = '[Link]
params = {'symbol': symbol, 'limit': 100}
resp = await self._http.get(url, params=params)
resp.raise_for_status()
data = [Link]()

rows = []
for item in data:
ts = [Link](item['fundingTime']/1000, tz=[Link])
rate = float(item['fundingRate'])
[Link]((ts, symbol, rate))

if rows:
# Compute Z-score vs rolling 30-day mean
rates = [r[2] for r in rows]
if len(rates) > 1:
mu = mean(rates)
sig = stdev(rates) or 0.0001
z = (rates[-1] - mu) / sig
else:
z = 0.0

async with self._pool.acquire() as conn:

await [Link](
'''INSERT INTO funding_rates
(time, symbol, funding_rate, source)
VALUES ($1, $2, $3, 'binance')
ON CONFLICT (time, symbol) DO NOTHING''',
[(r[0], r[1], r[2]) for r in rows],
)
[Link]('funding_stored', symbol=symbol,

SAS Phase 2 Implementation Guide · 2026 All 7 Deliverables · Full Production Code
SAS · PHASE 2 · DATA ACQUISITION & MARKET STATE FOUNDATION ROUND 1 · FOUNDATION

latest_rate=rates[-1], z_score=round(z,3))

# ── OPEN INTEREST ─────────────────────────────────────────────

async def _oi_loop(self) -> None:
while True:
try:
for symbol in SYMBOLS:
await self._fetch_and_store_oi(symbol)
except Exception as exc:
[Link]('oi_loop_error', error=str(exc))
await [Link](60 * 15) # 15 minutes

async def _fetch_and_store_oi(self, symbol: str) -> None:

url = '[Link]
resp = await self._http.get(url, params={'symbol': symbol})
resp.raise_for_status()
data = [Link]()

ts = [Link](data['time']/1000, tz=[Link])
oi = Decimal(data['openInterest'])

async with self._pool.acquire() as conn:

await [Link](
'''INSERT INTO open_interest (time, symbol, oi_value, source)
VALUES ($1, $2, $3, 'binance')
ON CONFLICT (time, symbol) DO NOTHING''',
ts, symbol, oi,
)
[Link]('oi_stored', symbol=symbol, oi=float(oi))

# ── LONG/SHORT RATIO (Coinglass) ──────────────────────────────

async def _ls_ratio_loop(self) -> None:
while True:
try:
for symbol in SYMBOLS:
await self._fetch_ls_ratio(symbol)
except Exception as exc:
[Link]('ls_ratio_error', error=str(exc))
await [Link](60 * 15) # 15 minutes

async def _fetch_ls_ratio(self, symbol: str) -> None:

if not settings.coinglass_api_key:
return # Skip if not configured
coin = 'BTC' if 'BTC' in symbol else 'ETH'
headers = {'coinglassSecret': settings.coinglass_api_key}
url = f'{COINGLASS_BASE}/indicator/long_short_account_ratio'
params = {'symbol': coin, 'interval': '15m', 'limit': 1}
try:
resp = await self._http.get(url, headers=headers, params=params,
timeout=10.0)
resp.raise_for_status()
# Store in system_events as JSON payload for Phase 5 FA scorer
data = [Link]()
async with self._pool.acquire() as conn:
await [Link](

SAS Phase 2 Implementation Guide · 2026 All 7 Deliverables · Full Production Code
SAS · PHASE 2 · DATA ACQUISITION & MARKET STATE FOUNDATION ROUND 1 · FOUNDATION

'''INSERT INTO system_events

(event_type, phase, severity, payload)
VALUES ('ls_ratio_update', 'P2', 'INFO', $1)''',
data,
)
except Exception as exc:
[Link]('ls_ratio_fetch_failed', symbol=symbol, error=str(exc))

# ── DERIVATIVES STATE BUILDER ─────────────────────────────────

async def get_current_state(self, symbol: str) -> DerivativesState | None:
"""Build current DerivativesState for MarketState assembly.

Queries the most recent values from TimescaleDB.

Marks as stale if data is older than freshness thresholds from D4.
"""
async with self._pool.acquire() as conn:
row = await [Link](
'''SELECT funding_rate, time FROM funding_rates
WHERE symbol=$1 ORDER BY time DESC LIMIT 1''',
symbol,
)
oi_row = await [Link](
'''SELECT oi_value, time FROM open_interest
WHERE symbol=$1 ORDER BY time DESC LIMIT 1''',
symbol,
)

if not row:
return None

now = [Link]([Link])
stale = (now - row['time']).total_seconds() > 8.5 * 3600

return DerivativesState(
time = row['time'],
symbol = symbol,
funding_rate = float(row['funding_rate']),
funding_z_score= 0.0, # Computed by Phase 5 FA scorer
open_interest = oi_row['oi_value'] if oi_row else Decimal(0),
oi_delta_4h = 0.0, # Computed by Phase 5
long_short_ratio = 1.0,
is_stale = stale,
)

SAS Phase 2 Implementation Guide · 2026 All 7 Deliverables · Full Production Code
SAS · PHASE 2 · DATA ACQUISITION & MARKET STATE FOUNDATION ROUND 1 · FOUNDATION

Sentiment & On-Chain Pipeline

D5 Fear & Greed · Glassnode · CryptoCompare NLP · Daily ingestion

The sentiment pipeline runs daily ingestion tasks for the Fear & Greed index, Glassnode on-chain metrics
(exchange net flow, SOPR, MVRV Z-score), and CryptoCompare news with NLP sentiment scoring. All data is
stored with freshness metadata so downstream phases can detect stale values per the D4 thresholds.

sas/data/[Link]
sas/data/[Link]
from __future__ import annotations

import asyncio
from datetime import datetime, timezone, timedelta

import asyncpg
import httpx

from [Link] import settings

from [Link] import get_logger
from [Link] import SentimentState

logger = get_logger(__name__)

ALTERNATIVE_ME_URL = '[Link]
GLASSNODE_BASE = '[Link]
CRYPTOCOMPARE_BASE = '[Link]

class SentimentPipeline:
"""Daily ingestion of sentiment and on-chain data.

Schedule:
- Fear & Greed index: every 12 hours (daily updates)
- Glassnode metrics: every 24 hours at 01:00 UTC
- News sentiment: every 2 hours

PIT Note: All daily data is stored with its publication date.
Phase 18 feature engineering uses T-1 values for intraday signals
(today's data is not yet final when an intraday trade is taken).
"""

def __init__(self, pool: [Link], http: [Link]) -> None:

self._pool = pool
self._http = http
self._tasks: list[[Link]] = []

async def start(self) -> None:

self._tasks.append(asyncio.create_task(
self._fear_greed_loop(), name='fear_greed'
))
self._tasks.append(asyncio.create_task(

SAS Phase 2 Implementation Guide · 2026 All 7 Deliverables · Full Production Code
SAS · PHASE 2 · DATA ACQUISITION & MARKET STATE FOUNDATION ROUND 1 · FOUNDATION

self._glassnode_loop(), name='glassnode'
))
self._tasks.append(asyncio.create_task(
self._news_loop(), name='news_sentiment'
))
[Link]('sentiment_pipeline_started')

async def stop(self) -> None:

for t in self._tasks: [Link]()
await [Link](*self._tasks, return_exceptions=True)

# ── FEAR & GREED ──────────────────────────────────────────────

async def _fear_greed_loop(self) -> None:
while True:
try:
await self._fetch_fear_greed()
except Exception as exc:
[Link]('fear_greed_error', error=str(exc))
await [Link](60 * 60 * 12) # Every 12 hours

async def _fetch_fear_greed(self) -> None:

resp = await self._http.get(ALTERNATIVE_ME_URL, timeout=10.0)
resp.raise_for_status()
data = [Link]()
entries = [Link]('data', [])
if not entries:
return

latest = entries[0]
value = int(latest['value'])
ts = [Link](int(latest['timestamp']), tz=[Link])

# Compute Z-score vs last 90 days of stored values

async with self._pool.acquire() as conn:
rows = await [Link](
'''SELECT payload->>'value' as v FROM system_events
WHERE event_type='fear_greed_update'
AND event_time >= NOW() - INTERVAL '90 days'
ORDER BY event_time DESC''',
)
historical = [float(r['v']) for r in rows if r['v']]

if len(historical) > 5:
from statistics import mean, stdev
mu = mean(historical)
sig = stdev(historical) or 1.0
z = (value - mu) / sig
else:
z = 0.0

await [Link](
'''INSERT INTO system_events
(event_type, phase, severity, payload)
VALUES ('fear_greed_update', 'P2', 'INFO', $1)''',
{'value': value, 'timestamp': [Link](), 'z_score': z},

SAS Phase 2 Implementation Guide · 2026 All 7 Deliverables · Full Production Code
SAS · PHASE 2 · DATA ACQUISITION & MARKET STATE FOUNDATION ROUND 1 · FOUNDATION

)
[Link]('fear_greed_stored', value=value, z_score=round(z,3))

# ── GLASSNODE ─────────────────────────────────────────────────
async def _glassnode_loop(self) -> None:
while True:
await [Link](
self._seconds_until_utc_hour(1) # Run at 01:00 UTC daily
)
if not settings.glassnode_api_key:
await [Link](3600)
continue
try:
await self._fetch_glassnode_metrics()
except Exception as exc:
[Link]('glassnode_error', error=str(exc))
await [Link](60 * 60 * 24) # Then every 24 hours

async def _fetch_glassnode_metrics(self) -> None:

metrics = {
'exchange_net_flow': '/transactions/transfers_volume_to_exchanges_sum',
'sopr': '/indicators/sopr',
'mvrv_z_score': '/market/mvrv_z_score',
}
headers = {'X-Api-Key': settings.glassnode_api_key}
for name, endpoint in [Link]():
try:
url = f'{GLASSNODE_BASE}{endpoint}'
params = {'a': 'BTC', 'i': '24h', 'f': 'JSON', 'limit': 7}
resp = await self._http.get(url, headers=headers,
params=params, timeout=15.0)
resp.raise_for_status()
data = [Link]()
if data:
latest = data[-1]
async with self._pool.acquire() as conn:
await [Link](
'''INSERT INTO system_events
(event_type, phase, severity, payload)
VALUES ($1, 'P2', 'INFO', $2)''',
f'glassnode_{name}',
{'t': [Link]('t'), 'v': [Link]('v')},
)
[Link]('glassnode_stored', metric=name)
except Exception as exc:
[Link]('glassnode_metric_failed', metric=name,
error=str(exc))

# ── NEWS SENTIMENT ────────────────────────────────────────────

async def _news_loop(self) -> None:
while True:
if settings.cryptocompare_api_key:
try:
await self._fetch_news_sentiment()
except Exception as exc:

SAS Phase 2 Implementation Guide · 2026 All 7 Deliverables · Full Production Code
SAS · PHASE 2 · DATA ACQUISITION & MARKET STATE FOUNDATION ROUND 1 · FOUNDATION

[Link]('news_sentiment_error', error=str(exc))
await [Link](60 * 60 * 2) # Every 2 hours

async def _fetch_news_sentiment(self) -> None:

headers = {'authorization': f'Apikey {settings.cryptocompare_api_key}'}
params = {'categories': 'BTC,ETH', 'sortOrder': 'latest', 'limit': 50}
resp = await self._http.get(CRYPTOCOMPARE_BASE, headers=headers,
params=params, timeout=15.0)
resp.raise_for_status()
articles = [Link]().get('Data', [])

# Simple sentiment: positive headlines +1, negative -1, neutral 0

# Phase 5 will apply proper NLP; this is the raw storage layer
POSITIVE = ['bullish','surge','rally','rise','gain','record','high','buy']
NEGATIVE = ['bearish','crash','drop','fall','loss','low','fear','sell']

scores = []
for article in articles:
title = [Link]('title','').lower()
pos = sum(1 for w in POSITIVE if w in title)
neg = sum(1 for w in NEGATIVE if w in title)
score = (pos - neg) / max(pos + neg, 1)
[Link](score)

avg_sentiment = sum(scores) / len(scores) if scores else 0.5

# Normalise to 0.0–1.0 range
normalised = (avg_sentiment + 1) / 2

async with self._pool.acquire() as conn:

await [Link](
'''INSERT INTO system_events
(event_type, phase, severity, payload)
VALUES ('news_sentiment_update', 'P2', 'INFO', $1)''',
{'score': normalised, 'article_count': len(articles),
'computed_at': [Link]([Link]).isoformat()},
)
[Link]('news_sentiment_stored', score=round(normalised,3))

@staticmethod
def _seconds_until_utc_hour(target_hour: int) -> float:
now = [Link]([Link])
target = [Link](hour=target_hour, minute=0, second=0, microsecond=0)
if target <= now:
target += timedelta(days=1)
return (target - now).total_seconds()

SAS Phase 2 Implementation Guide · 2026 All 7 Deliverables · Full Production Code
SAS · PHASE 2 · DATA ACQUISITION & MARKET STATE FOUNDATION ROUND 1 · FOUNDATION

Data Quality Monitor & Auto-Backfill

D6 Gap detection · Auto-backfill · data_quality_events · DLQ alerts

The data quality monitor runs every 5 minutes, checking all critical data streams for gaps, staleness, and
latency breaches. Every anomaly is written to the data_quality_events table (C6). On OHLCV gap detection, it
automatically triggers a targeted backfill. On 3+ consecutive failures, it publishes a DataGapDetectedEvent to
the event bus.

sas/data/quality_monitor.py
sas/data/quality_monitor.py
from __future__ import annotations

import asyncio
from datetime import datetime, timezone, timedelta
from typing import TYPE_CHECKING

import asyncpg

from [Link] import get_logger

from [Link] import bus, DataGapDetectedEvent

if TYPE_CHECKING:
from [Link].binance_rest import BinanceRestConnector

logger = get_logger(__name__)

SYMBOLS = ['BTCUSDT', 'ETHUSDT']

TIMEFRAMES = ['1m', '5m', '15m', '1h', '4h', '1d']

# Staleness thresholds (seconds) — from Phase 0 D4

STALE_OHLCV_SEC = 65 # 1 candle + 5s grace
STALE_OB_SEC = 10 # Order book
STALE_FUNDING_SEC = 28800 # 8 hours
STALE_OI_SEC = 900 # 15 minutes

# Timeframe durations in seconds

TF_SECONDS = {
'1m':60,'5m':300,'15m':900,'1h':3600,'4h':14400,'1d':86400
}

class DataQualityMonitor:
"""Runs every 5 minutes — checks all data streams for gaps and staleness.

On any anomaly:
1. Writes event to data_quality_events table
2. Triggers auto-backfill for OHLCV gaps
3. After 3 consecutive failures: publishes DataGapDetectedEvent
(Phase 9 may halt execution if this breaches latency threshold)
"""

SAS Phase 2 Implementation Guide · 2026 All 7 Deliverables · Full Production Code
SAS · PHASE 2 · DATA ACQUISITION & MARKET STATE FOUNDATION ROUND 1 · FOUNDATION

def __init__(
self,
pool: [Link],
rest_connector: 'BinanceRestConnector',
) -> None:
self._pool = pool
self._rest = rest_connector
self._failures: dict[str, int] = {} # key -> consecutive failure count
self._task: [Link] | None = None

async def start(self) -> None:

self._task = asyncio.create_task(
self._monitor_loop(), name='data_quality_monitor'
)
[Link]('quality_monitor_started')

async def stop(self) -> None:

if self._task:
self._task.cancel()
await [Link](self._task, return_exceptions=True)

async def _monitor_loop(self) -> None:

while True:
try:
await self._run_checks()
except Exception as exc:
[Link]('quality_monitor_error', error=str(exc))
await [Link](300) # Every 5 minutes

async def _run_checks(self) -> None:

for symbol in SYMBOLS:
for tf in TIMEFRAMES:
await self._check_ohlcv_gap(symbol, tf)
await self._check_derivatives_freshness(symbol)

# ── OHLCV GAP DETECTION ───────────────────────────────────────

async def _check_ohlcv_gap(self, symbol: str, tf: str) -> None:
key = f'ohlcv:{symbol}:{tf}'
tf_sec = TF_SECONDS[tf]
expected_max_age = tf_sec + 65 # One full candle + buffer

async with self._pool.acquire() as conn:

row = await [Link](
'''SELECT MAX(time) as last_time FROM ohlcv
WHERE symbol=$1 AND timeframe=$2''',
symbol, tf,
)

if not row or not row['last_time']:

await self._record_anomaly(
key, 'missing_data',
f'No OHLCV data for {symbol} {tf}',
)
return

SAS Phase 2 Implementation Guide · 2026 All 7 Deliverables · Full Production Code
SAS · PHASE 2 · DATA ACQUISITION & MARKET STATE FOUNDATION ROUND 1 · FOUNDATION

age = ([Link]([Link]) - row['last_time']).total_seconds()

if age > expected_max_age:

gap_start = row['last_time']
gap_end = [Link]([Link])
await self._record_anomaly(
key, 'ohlcv_gap',
f'{symbol} {tf} gap: last candle {age:.0f}s ago',
auto_resolve=True,
)
# Trigger auto-backfill
[Link]('auto_backfill_triggered', symbol=symbol, tf=tf)
filled = await self._rest.fill_gap(symbol, tf, gap_start, gap_end)
await self._mark_resolved(key, f'Backfilled {filled} rows')
self._failures[key] = 0 # Reset on successful resolution
else:
self._failures[key] = 0 # Clear failure count on pass

# ── DERIVATIVES FRESHNESS ─────────────────────────────────────

async def _check_derivatives_freshness(self, symbol: str) -> None:
key = f'funding:{symbol}'
async with self._pool.acquire() as conn:
row = await [Link](
'SELECT MAX(time) as t FROM funding_rates WHERE symbol=$1',
symbol,
)
if row and row['t']:
age = ([Link]([Link]) - row['t']).total_seconds()
if age > STALE_FUNDING_SEC:
await self._record_anomaly(key, 'stale_funding',
f'{symbol} funding rate {age/3600:.1f}h old')

# ── ANOMALY RECORDING ─────────────────────────────────────────

async def _record_anomaly(
self, key: str, event_type: str,
description: str, auto_resolve: bool = False,
) -> None:
self._failures[key] = self._failures.get(key, 0) + 1
count = self._failures[key]

[Link]('data_quality_anomaly', key=key,
event_type=event_type, consecutive=count,
description=description)

async with self._pool.acquire() as conn:

await [Link](
'''INSERT INTO data_quality_events
(source, event_type, description)
VALUES ($1, $2, $3)''',
key, event_type, description,
)

# After 3 consecutive failures: publish to event bus

# Phase 9 monitoring listens for DataGapDetectedEvent
if count >= 3:

SAS Phase 2 Implementation Guide · 2026 All 7 Deliverables · Full Production Code
SAS · PHASE 2 · DATA ACQUISITION & MARKET STATE FOUNDATION ROUND 1 · FOUNDATION

await [Link](DataGapDetectedEvent(
source=key,
description=f'3+ consecutive failures: {description}',
))
[Link]('data_quality_critical', key=key, count=count)

async def _mark_resolved(self, key: str, resolution: str) -> None:

async with self._pool.acquire() as conn:
await [Link](
'''UPDATE data_quality_events
SET resolved_at=$1, resolution=$2
WHERE source=$3 AND resolved_at IS NULL
ORDER BY detected_at DESC LIMIT 1''',
[Link]([Link]), resolution, key,
)

SAS Phase 2 Implementation Guide · 2026 All 7 Deliverables · Full Production Code
SAS · PHASE 2 · DATA ACQUISITION & MARKET STATE FOUNDATION ROUND 1 · FOUNDATION

Replay Engine
D7 Zero look-ahead MarketState iterator · Sole backtest data source

The Replay Engine is the most architecturally critical component in Phase 2. It is the sole data source for all
Phase 7 backtesting. Its defining guarantee: when replaying at timestamp T, it yields only data that was
available at timestamp T. This zero look-ahead guarantee is enforced by strict WHERE clause filtering — no
LEAD() functions, no future-looking window functions, no post-hoc corrections.

Why This Matters More Than Anything Else in Phase 2

A Replay Engine with even subtle look-ahead bias will make Phase 7 backtests appear significantly better than
live performance will ever be. This leads to overconfident capital deployment at a tier that the strategy has not
actually earned. The WHERE clause 'time <= $snapshot_time' must be applied to every single data query. No
exceptions. Test this with a dedicated anti-lookahead test suite.

sas/data/replay_engine.py
sas/data/replay_engine.py
from __future__ import annotations

from datetime import datetime, timezone

from decimal import Decimal
from typing import AsyncIterator

import asyncpg

from [Link] import get_logger

from [Link] import (
MarketState, OHLCVBar, OrderBookState, DerivativesState, SentimentState
)

logger = get_logger(__name__)

CANDLE_LOOKBACK = {
'1m': 200, # bars of history for each timeframe
'5m': 200,
'15m': 200,
'1h': 200,
'4h': 200,
'1d': 200,
}

class ReplayEngine:
"""Yields MarketState objects for backtesting.

ZERO LOOK-AHEAD GUARANTEE:

Every database query uses WHERE time <= snapshot_time.
No data available after snapshot_time is ever returned.
This guarantee must hold for ALL data types:
- OHLCV candles

SAS Phase 2 Implementation Guide · 2026 All 7 Deliverables · Full Production Code
SAS · PHASE 2 · DATA ACQUISITION & MARKET STATE FOUNDATION ROUND 1 · FOUNDATION

- Order book snapshots

- Funding rates and open interest
- Fear & Greed index
- Glassnode metrics (T-1 only for daily data)
- News sentiment (rolling window capped at snapshot_time)

This is the ONLY data source used by Phase 7 backtesting.

It is never replaced by a simplified simulation.
"""

def __init__(self, pool: [Link]) -> None:

self._pool = pool

async def replay(

self,
symbol: str,
timeframe: str,
start_time: datetime,
end_time: datetime,
) -> AsyncIterator[MarketState]:
"""Iterate over MarketState objects in chronological order.

Each yielded MarketState represents the complete market view

available at that candle's close time.

Usage:
async for state in [Link]('BTCUSDT','4h', start, end):
# [Link] is the candle close time
# all state data is available as-of [Link]
context = market_structure_engine.evaluate(state)
"""
# Fetch all candle close times in range
async with self._pool.acquire() as conn:
timestamps = await [Link](
'''SELECT DISTINCT time FROM ohlcv
WHERE symbol=$1 AND timeframe=$2
AND time >= $3 AND time <= $4
ORDER BY time ASC''',
symbol, timeframe, start_time, end_time,
)

[Link]('replay_starting',
symbol=symbol, timeframe=timeframe,
candles=len(timestamps))

for row in timestamps:

snapshot_time = row['time']
state = await self._build_state(symbol, snapshot_time)
yield state

async def _build_state(

self, symbol: str, snapshot_time: datetime
) -> MarketState:
"""Build a complete MarketState as-of snapshot_time.

SAS Phase 2 Implementation Guide · 2026 All 7 Deliverables · Full Production Code
SAS · PHASE 2 · DATA ACQUISITION & MARKET STATE FOUNDATION ROUND 1 · FOUNDATION

ALL queries use WHERE time <= snapshot_time.

This is the zero look-ahead guarantee.
"""
async with self._pool.acquire() as conn:
candles = await self._fetch_candles(conn, symbol, snapshot_time)
ob = await self._fetch_orderbook(conn, symbol, snapshot_time)
deriv = await self._fetch_derivatives(conn, symbol, snapshot_time)
sent = await self._fetch_sentiment(conn, snapshot_time)
atr = self._compute_atr(candles)

return MarketState(
timestamp = snapshot_time,
symbol = symbol,
candles = candles,
order_book = ob,
derivatives = deriv,
sentiment = sent,
atr = atr,
)

async def _fetch_candles(

self, conn: [Link],
symbol: str, snapshot_time: datetime,
) -> dict[str, list[OHLCVBar]]:
result: dict[str, list[OHLCVBar]] = {}
for tf, lookback in CANDLE_LOOKBACK.items():
rows = await [Link](
'''SELECT time, open, high, low, close, volume FROM ohlcv
WHERE symbol=$1 AND timeframe=$2
AND time <= $3
ORDER BY time DESC LIMIT $4''',
symbol, tf, snapshot_time, lookback,
)
result[tf] = [
OHLCVBar(
time=r['time'], symbol=symbol, timeframe=tf,
open=r['open'], high=r['high'],
low=r['low'], close=r['close'], volume=r['volume'],
)
for r in reversed(rows) # Chronological order
]
return result

async def _fetch_orderbook(

self, conn: [Link],
symbol: str, snapshot_time: datetime,
) -> OrderBookState | None:
row = await [Link](
'''SELECT * FROM orderbook_snapshots
WHERE symbol=$1 AND time <= $2
ORDER BY time DESC LIMIT 1''',
symbol, snapshot_time,
)
if not row:
return None

SAS Phase 2 Implementation Guide · 2026 All 7 Deliverables · Full Production Code
SAS · PHASE 2 · DATA ACQUISITION & MARKET STATE FOUNDATION ROUND 1 · FOUNDATION

return OrderBookState(
time=row['time'], symbol=symbol,
bid_volume_top5=row['bid_volume_top5'],
ask_volume_top5=row['ask_volume_top5'],
imbalance_ratio=float(row['imbalance_ratio']),
best_bid=row['best_bid'], best_ask=row['best_ask'],
spread_pct=float(row['spread_pct']),
)

async def _fetch_derivatives(

self, conn: [Link],
symbol: str, snapshot_time: datetime,
) -> DerivativesState | None:
row = await [Link](
'''SELECT * FROM funding_rates
WHERE symbol=$1 AND time <= $2
ORDER BY time DESC LIMIT 1''',
symbol, snapshot_time,
)
if not row:
return None
return DerivativesState(
time=row['time'], symbol=symbol,
funding_rate=float(row['funding_rate']),
funding_z_score=0.0, open_interest=Decimal(0),
oi_delta_4h=0.0, long_short_ratio=1.0,
)

async def _fetch_sentiment(

self, conn: [Link],
snapshot_time: datetime,
) -> SentimentState | None:
# Daily data: use T-1 — data published 'today' was not
# available at intraday signal times
t_minus_1 = snapshot_time.replace(
hour=0, minute=0, second=0, microsecond=0
)
row = await [Link](
'''SELECT payload FROM system_events
WHERE event_type='fear_greed_update'
AND event_time <= $1
ORDER BY event_time DESC LIMIT 1''',
t_minus_1,
)
if not row:
return None
payload = row['payload']
return SentimentState(
time=snapshot_time,
fear_greed_index=int([Link]('value', 50)),
fear_greed_z=float([Link]('z_score', 0.0)),
exchange_net_flow=0.0,
news_sentiment=0.5,
)

SAS Phase 2 Implementation Guide · 2026 All 7 Deliverables · Full Production Code
SAS · PHASE 2 · DATA ACQUISITION & MARKET STATE FOUNDATION ROUND 1 · FOUNDATION

@staticmethod
def _compute_atr(
candles: dict[str, list[OHLCVBar]],
period: int = 14,
) -> dict[str, float]:
atr: dict[str, float] = {}
for tf, bars in [Link]():
if len(bars) < 2:
atr[tf] = 0.0
continue
trs = []
for i in range(1, min(period + 1, len(bars))):
high = float(bars[i].high)
low = float(bars[i].low)
prev_close = float(bars[i-1].close)
tr = max(high - low,
abs(high - prev_close),
abs(low - prev_close))
[Link](tr)
atr[tf] = sum(trs) / len(trs) if trs else 0.0
return atr

tests/unit/test_replay_engine.py — Anti-Look-Ahead Tests

tests/unit/test_replay_engine.py
"""Critical tests: verify zero look-ahead guarantee of the Replay Engine.

These tests are the most important in the entire Phase 2 test suite.
A Replay Engine with look-ahead bias invalidates every backtest result.
"""
import pytest
from datetime import datetime, timezone, timedelta
from [Link] import AsyncMock, MagicMock

from [Link].replay_engine import ReplayEngine

class TestZeroLookAhead:
async def test_candles_strictly_before_snapshot_time(self, mock_pool):
"""No candle with time > snapshot_time must ever appear in state."""
snapshot_time = datetime(2025, 6, 15, 12, 0, tzinfo=[Link])
engine = ReplayEngine(mock_pool)

# Inject a candle that is AFTER snapshot_time

future_candle_time = snapshot_time + timedelta(hours=4)
mock_pool.acquire().__aenter__().fetchrow.return_value = {
'time': future_candle_time, 'open': 50000, 'high': 51000,
'low': 49000, 'close': 50500, 'volume': 100,
}

state = await engine._build_state('BTCUSDT', snapshot_time)

# All candles must have time <= snapshot_time

SAS Phase 2 Implementation Guide · 2026 All 7 Deliverables · Full Production Code
SAS · PHASE 2 · DATA ACQUISITION & MARKET STATE FOUNDATION ROUND 1 · FOUNDATION

for tf, bars in [Link]():

for bar in bars:
assert [Link] <= snapshot_time, (
f'LOOK-AHEAD BIAS: {tf} candle at {[Link]} '
f'is after snapshot {snapshot_time}'
)

async def test_orderbook_strictly_before_snapshot(self, mock_pool):

"""Order book snapshot must be <= snapshot_time."""
snapshot_time = datetime(2025, 6, 15, 12, 0, tzinfo=[Link])
# This test verifies the WHERE time <= $2 clause is present
# in _fetch_orderbook by checking query args passed to the mock
engine = ReplayEngine(mock_pool)
await engine._fetch_orderbook(
mock_pool.acquire().__aenter__(),
'BTCUSDT',
snapshot_time,
)
call_args = mock_pool.acquire().__aenter__().fetchrow.call_args
query = call_args[0][0]
assert 'time <= $2' in query or 'time<=$2' in query, (
'ORDER BOOK QUERY MISSING look-ahead guard: time <= snapshot_time'
)

async def test_sentiment_uses_t_minus_1(self, mock_pool):

"""Fear & Greed must use T-1 date boundary, not snapshot_time."""
snapshot_time = datetime(2025, 6, 15, 14, 30, tzinfo=[Link])
engine = ReplayEngine(mock_pool)
await engine._fetch_sentiment(
mock_pool.acquire().__aenter__(),
snapshot_time,
)
call_args = mock_pool.acquire().__aenter__().fetchrow.call_args
# The T-1 boundary passed to the query must be midnight,
# not 14:30 (the snapshot time)
passed_time = call_args[0][1]
assert passed_time.hour == 0 and passed_time.minute == 0, (
'SENTIMENT look-ahead: must use T-1 midnight, not snapshot_time'
)

SAS Phase 2 Implementation Guide · 2026 All 7 Deliverables · Full Production Code
SAS · PHASE 2 · DATA ACQUISITION & MARKET STATE FOUNDATION ROUND 1 · FOUNDATION

MarketState Publisher — Wiring It All Together

The MarketState publisher assembles all Phase 2 components into a single pipeline that publishes a complete
MarketState to Redis on every candle close. This is the object that Phase 3 and Phase 4 will consume.

sas/data/market_state_publisher.py
sas/data/market_state_publisher.py
from __future__ import annotations

import asyncio
import json
from datetime import datetime, timezone

import asyncpg
import [Link] as aioredis

from [Link] import get_logger

from [Link] import MarketState
from [Link] import OrderBookProcessor
from [Link] import DerivativesPipeline
from [Link] import SentimentPipeline
from [Link].replay_engine import ReplayEngine

logger = get_logger(__name__)

# Redis key pattern: 'ms:{symbol}:{timeframe}'

MS_TTL = 120 # MarketState expires after 2 candle periods

class MarketStatePublisher:
"""Assembles and publishes MarketState on every candle close.

Called by the WebSocket kline handler when a candle closes.

Publishes to Redis for zero-latency consumption by Phase 3/4.
Also persists the state reference to TimescaleDB for replay.
"""

def __init__(
self,
pool: [Link],
redis: [Link],
ob_processor: OrderBookProcessor,
derivatives: DerivativesPipeline,
) -> None:
self._pool = pool
self._redis = redis
self._ob = ob_processor
self._deriv = derivatives

async def on_kline_close(self, kline_data: dict) -> None:

"""Called by WebSocket manager when a kline closes (is_closed=True)."""

SAS Phase 2 Implementation Guide · 2026 All 7 Deliverables · Full Production Code
SAS · PHASE 2 · DATA ACQUISITION & MARKET STATE FOUNDATION ROUND 1 · FOUNDATION

kline = kline_data.get('k', {})

if not [Link]('x', False):
return # Candle not yet closed

symbol = [Link]('s', '') # e.g. 'BTCUSDT'

timeframe = [Link]('i', '') # e.g. '1m'
close_time = [Link](
[Link]('T', 0) / 1000, tz=[Link]
)

try:
state = await self._build(symbol, timeframe, close_time)
await self._publish_to_redis(state)
[Link]('market_state_published',
symbol=symbol, timeframe=timeframe,
close_time=str(close_time))
except Exception as exc:
[Link]('market_state_build_failed',
symbol=symbol, tf=timeframe, error=str(exc))

async def _build(

self, symbol: str, tf: str, ts: datetime
) -> MarketState:
async with self._pool.acquire() as conn:
rows = await [Link](
'''SELECT time,open,high,low,close,volume FROM ohlcv
WHERE symbol=$1 AND timeframe=$2
AND time <= $3
ORDER BY time DESC LIMIT 200''',
symbol, tf, ts,
)
from [Link] import OHLCVBar
bars = [OHLCVBar(time=r['time'],symbol=symbol,timeframe=tf,
open=r['open'],high=r['high'],low=r['low'],
close=r['close'],volume=r['volume'])
for r in reversed(rows)]
ob = await self._ob.get_current(symbol)
deriv = await self._deriv.get_current_state(symbol)
from [Link].replay_engine import ReplayEngine
atr = ReplayEngine._compute_atr({tf: bars})
return MarketState(
timestamp=ts, symbol=symbol,
candles={tf: bars}, order_book=ob,
derivatives=deriv, atr=atr,
)

async def _publish_to_redis(

self, state: MarketState
) -> None:
key = f'ms:{[Link]}:{list([Link]())[0]}'
# Store as JSON — Phase 3 reads with [Link]()
payload = {
'timestamp': [Link](),
'symbol': [Link],
'candle_count': {k: len(v) for k,v in [Link]()},

SAS Phase 2 Implementation Guide · 2026 All 7 Deliverables · Full Production Code
SAS · PHASE 2 · DATA ACQUISITION & MARKET STATE FOUNDATION ROUND 1 · FOUNDATION

'ob_fresh': state.order_book is not None and not state.order_book.is_stale,

'atr': [Link],
}
await self._redis.setex(key, MS_TTL, [Link](payload))

Phase 2 Completion Checklist

Do not proceed to Phase 3 until every row is checked

Phase 3 (Market Structure Engine) will process MarketState objects on every candle close. If any Phase 2
component is incomplete or has subtle bugs, Phase 3 will silently produce incorrect structure analyses. There is
no error message for 'wrong candles in MarketState' — only wrong signals months later.

C Task Command to Verify

h
e
c
k

☐ 2-year OHLCV backfill complete for BTC + SELECT COUNT(*), timeframe FROM ohlcv GROUP BY timeframe
ETH, all 6 timeframes → ~500k+ rows per TF

☐ Checkpoint file shows all 12 combinations cat /app/data/.backfill_checkpoint.json → 12 keys all 'complete'
'complete'

☐ WebSocket connects and streams live klines docker logs sas-app | grep websocket_connected → appears on
without disconnect startup

☐ Order book snapshots appearing in SELECT COUNT(*) FROM orderbook_snapshots WHERE time >
TimescaleDB every ~100ms NOW() - INTERVAL '1 minute' → > 0

☐ Redis ob:BTCUSDT key returns fresh docker exec sas-cache redis-cli get ob:BTCUSDT → JSON with
imbalance_ratio imbalance_ratio field

☐ Redis ms:BTCUSDT:1m key updates on docker exec sas-cache redis-cli get ms:BTCUSDT:1m → JSON
every 1M candle close with recent timestamp

☐ Funding rates ingesting every 8 hours SELECT COUNT(*), MAX(time) FROM funding_rates → has recent
rows

☐ Open interest ingesting every 15 minutes SELECT COUNT(*), MAX(time) FROM open_interest → has recent
rows

☐ Fear & Greed index has at least 7 days of SELECT COUNT(*) FROM system_events WHERE
historical data event_type='fear_greed_update' → > 7

☐ Data quality monitor running and logging docker logs sas-app | grep quality_monitor → periodic log entries
every 5 minutes

☐ A simulated OHLCV gap triggers auto-backfill Run integration test: pytest tests/integration/test_quality_monitor.py -
v

☐ Replay Engine anti-look-ahead tests pass pytest tests/unit/test_replay_engine.py -v → ALL 3 look-ahead tests

SAS Phase 2 Implementation Guide · 2026 All 7 Deliverables · Full Production Code
SAS · PHASE 2 · DATA ACQUISITION & MARKET STATE FOUNDATION ROUND 1 · FOUNDATION

C Task Command to Verify

h
e
c
k

PASS

☐ Replay Engine yields MarketState in pytest tests/integration/test_replay_engine.py -v → all assertions

chronological order pass

☐ No data_quality_events with unresolved gaps SELECT * FROM data_quality_events WHERE resolved_at IS NULL
after 24h operation → empty

☐ CI pipeline green after all Phase 2 code GitHub Actions → latest push → all jobs green
added

☐ Phase 2 hours logged Actual hours: ______ (estimate was 175–250)

What Phase 3 Builds On Top Of

Phase 3 Component Phase 2 Dependency Correctness Requirement

Market Structure Engine [Link] — dict of Candles must be in chronological order, no duplicates,
OHLCVBar lists per timeframe no gaps

BOS / CHoCH Detection [Link]['4h'] and 200+ bars of 4H and 1D history required for reliable
['1d'] lookback bars swing detection

Order Block Identification [Link] — last N Candles must represent exact historical sequence —
bars before a BOS no reordering

Scalp OB Imbalance MarketState.order_book.imbalanc Must be < 10 seconds old. Stale OB = invalid scalp
Check e_ratio entry condition

ATR for SL Calculation [Link] — pre-computed ATR must use only candles available at
ATR per timeframe snapshot_time (no look-ahead)

Replay (Phase 7) [Link]() — the sole Zero look-ahead tests must be passing. Any failure
backtest data source invalidates all backtests.

Phase 2 Sign-Off
Phase 2 Start Date Phase 2 End Date Actual Hours Spent

________________ ________________ _______ hours

All 7 deliverables complete and verified:

Backfill complete · WebSocket live · Replay anti-lookahead tests passing Operator Signature: _____________
· CI green

SAS TRADING SYSTEM · PHASE 2 · DATA ACQUISITION & MARKET STATE · v1.0 · 2026

SAS Phase 2 Implementation Guide · 2026 All 7 Deliverables · Full Production Code
SAS · PHASE 2 · DATA ACQUISITION & MARKET STATE FOUNDATION ROUND 1 · FOUNDATION

PERSONAL USE ONLY · NOT FOR COMMERCIAL DISTRIBUTION

SAS Phase 2 Implementation Guide · 2026 All 7 Deliverables · Full Production Code