Adriano/Cerbero-Bite
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
589d003e5c20e6424f55f1a3fee2836b60fe7293
Cerbero-Bite/src/cerbero_bite/state/repository.py
T
root 19695e4730 feat(state): dvol_history multi-asset (ETH+BTC) + backfill ETH legacy rows 
Migration 0006 promuove dvol_history da PK=(timestamp) mono-ETH a
PK=(timestamp, asset), rinomina eth_spot -> spot, e backfilla con
asset='ETH' le righe storiche. market_snapshot_cycle ora scrive sia
per ETH che per BTC; monitor_cycle resta ETH-only via WHERE asset='ETH'
nella lookup di return_4h.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-05-12 13:38:34 +00:00

936 lines
33 KiB
Python
Raw Blame History

"""Typed CRUD layer over the SQLite database.
All methods take a :class:`sqlite3.Connection` so callers can compose a
single transaction across multiple writes (the orchestrator does this
when persisting an entry decision + the resulting proposal). The
repository never opens its own connection: that responsibility is left
to :func:`cerbero_bite.state.db.connect`.
Decimals are stored as TEXT to preserve precision (see
``state/models.py``).
"""
from __future__ import annotations
import sqlite3
from datetime import UTC, datetime, timedelta
from decimal import Decimal
from typing import Any
from uuid import UUID
from cerbero_bite.state.models import (
DecisionRecord,
DvolSnapshot,
InstructionRecord,
ManualAction,
MarketSnapshotRecord,
OptionChainQuoteRecord,
PositionRecord,
PositionStatus,
SystemStateRecord,
)
__all__ = ["Repository"]
# ---------------------------------------------------------------------------
# Encoding helpers
# ---------------------------------------------------------------------------
def _enc_dec(value: Decimal | None) -> str | None:
return None if value is None else str(value)
def _enc_dt(value: datetime | None) -> str | None:
if value is None:
return None
if value.tzinfo is None:
value = value.replace(tzinfo=UTC)
return value.astimezone(UTC).isoformat()
def _enc_uuid(value: UUID | None) -> str | None:
return None if value is None else str(value)
def _dec_dec(value: Any) -> Decimal | None:
if value is None:
return None
return Decimal(str(value))
def _dec_dt(value: Any) -> datetime | None:
if value is None:
return None
return datetime.fromisoformat(str(value))
def _dec_uuid(value: Any) -> UUID | None:
if value is None:
return None
return UUID(str(value))
# ---------------------------------------------------------------------------
# Repository
# ---------------------------------------------------------------------------
class Repository:
"""Typed CRUD wrapper. One instance per process, all calls take a conn."""
# ------------------------------------------------------------------
# positions
# ------------------------------------------------------------------
def create_position(self, conn: sqlite3.Connection, record: PositionRecord) -> None:
conn.execute(
"""
INSERT INTO positions (
proposal_id, spread_type, asset, expiry,
short_strike, long_strike, short_instrument, long_instrument,
n_contracts, spread_width_usd, spread_width_pct,
credit_eth, credit_usd, max_loss_usd,
spot_at_entry, dvol_at_entry, delta_at_entry, eth_price_at_entry,
proposed_at, opened_at, closed_at, close_reason,
debit_paid_eth, pnl_eth, pnl_usd,
status, created_at, updated_at
) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)
""",
(
_enc_uuid(record.proposal_id),
record.spread_type,
record.asset,
_enc_dt(record.expiry),
_enc_dec(record.short_strike),
_enc_dec(record.long_strike),
record.short_instrument,
record.long_instrument,
record.n_contracts,
_enc_dec(record.spread_width_usd),
_enc_dec(record.spread_width_pct),
_enc_dec(record.credit_eth),
_enc_dec(record.credit_usd),
_enc_dec(record.max_loss_usd),
_enc_dec(record.spot_at_entry),
_enc_dec(record.dvol_at_entry),
_enc_dec(record.delta_at_entry),
_enc_dec(record.eth_price_at_entry),
_enc_dt(record.proposed_at),
_enc_dt(record.opened_at),
_enc_dt(record.closed_at),
record.close_reason,
_enc_dec(record.debit_paid_eth),
_enc_dec(record.pnl_eth),
_enc_dec(record.pnl_usd),
record.status,
_enc_dt(record.created_at),
_enc_dt(record.updated_at),
),
)
def get_position(
self, conn: sqlite3.Connection, proposal_id: UUID
) -> PositionRecord | None:
row = conn.execute(
"SELECT * FROM positions WHERE proposal_id = ?",
(_enc_uuid(proposal_id),),
).fetchone()
return None if row is None else _row_to_position(row)
def list_positions(
self,
conn: sqlite3.Connection,
*,
status: PositionStatus | None = None,
) -> list[PositionRecord]:
if status is None:
cursor = conn.execute(
"SELECT * FROM positions ORDER BY created_at DESC"
)
else:
cursor = conn.execute(
"SELECT * FROM positions WHERE status = ? ORDER BY created_at DESC",
(status,),
)
return [_row_to_position(r) for r in cursor.fetchall()]
def list_open_positions(self, conn: sqlite3.Connection) -> list[PositionRecord]:
rows = conn.execute(
"SELECT * FROM positions WHERE status IN ('open', 'awaiting_fill', "
"'closing') ORDER BY created_at DESC"
).fetchall()
return [_row_to_position(r) for r in rows]
def update_position_status(
self,
conn: sqlite3.Connection,
proposal_id: UUID,
*,
status: PositionStatus,
opened_at: datetime | None = None,
closed_at: datetime | None = None,
close_reason: str | None = None,
debit_paid_eth: Decimal | None = None,
pnl_eth: Decimal | None = None,
pnl_usd: Decimal | None = None,
now: datetime,
) -> None:
sets: list[str] = ["status = ?", "updated_at = ?"]
params: list[Any] = [status, _enc_dt(now)]
if opened_at is not None:
sets.append("opened_at = ?")
params.append(_enc_dt(opened_at))
if closed_at is not None:
sets.append("closed_at = ?")
params.append(_enc_dt(closed_at))
if close_reason is not None:
sets.append("close_reason = ?")
params.append(close_reason)
if debit_paid_eth is not None:
sets.append("debit_paid_eth = ?")
params.append(_enc_dec(debit_paid_eth))
if pnl_eth is not None:
sets.append("pnl_eth = ?")
params.append(_enc_dec(pnl_eth))
if pnl_usd is not None:
sets.append("pnl_usd = ?")
params.append(_enc_dec(pnl_usd))
params.append(_enc_uuid(proposal_id))
conn.execute(
f"UPDATE positions SET {', '.join(sets)} WHERE proposal_id = ?",
params,
)
def count_concurrent_positions(self, conn: sqlite3.Connection) -> int:
row = conn.execute(
"SELECT COUNT(*) FROM positions WHERE status IN "
"('awaiting_fill', 'open', 'closing')"
).fetchone()
return int(row[0])
# ------------------------------------------------------------------
# instructions
# ------------------------------------------------------------------
def create_instruction(
self, conn: sqlite3.Connection, record: InstructionRecord
) -> None:
conn.execute(
"""
INSERT INTO instructions (
instruction_id, proposal_id, kind, payload_json,
sent_at, acknowledged_at, filled_at, cancelled_at,
actual_fill_eth, actual_fees_eth
) VALUES (?,?,?,?,?,?,?,?,?,?)
""",
(
_enc_uuid(record.instruction_id),
_enc_uuid(record.proposal_id),
record.kind,
record.payload_json,
_enc_dt(record.sent_at),
_enc_dt(record.acknowledged_at),
_enc_dt(record.filled_at),
_enc_dt(record.cancelled_at),
_enc_dec(record.actual_fill_eth),
_enc_dec(record.actual_fees_eth),
),
)
def update_instruction(
self,
conn: sqlite3.Connection,
instruction_id: UUID,
*,
acknowledged_at: datetime | None = None,
filled_at: datetime | None = None,
cancelled_at: datetime | None = None,
actual_fill_eth: Decimal | None = None,
actual_fees_eth: Decimal | None = None,
) -> None:
sets: list[str] = []
params: list[Any] = []
if acknowledged_at is not None:
sets.append("acknowledged_at = ?")
params.append(_enc_dt(acknowledged_at))
if filled_at is not None:
sets.append("filled_at = ?")
params.append(_enc_dt(filled_at))
if cancelled_at is not None:
sets.append("cancelled_at = ?")
params.append(_enc_dt(cancelled_at))
if actual_fill_eth is not None:
sets.append("actual_fill_eth = ?")
params.append(_enc_dec(actual_fill_eth))
if actual_fees_eth is not None:
sets.append("actual_fees_eth = ?")
params.append(_enc_dec(actual_fees_eth))
if not sets:
return
params.append(_enc_uuid(instruction_id))
conn.execute(
f"UPDATE instructions SET {', '.join(sets)} "
f"WHERE instruction_id = ?",
params,
)
def list_instructions(
self, conn: sqlite3.Connection, proposal_id: UUID
) -> list[InstructionRecord]:
rows = conn.execute(
"SELECT * FROM instructions WHERE proposal_id = ? ORDER BY sent_at",
(_enc_uuid(proposal_id),),
).fetchall()
return [_row_to_instruction(r) for r in rows]
# ------------------------------------------------------------------
# decisions
# ------------------------------------------------------------------
def record_decision(
self, conn: sqlite3.Connection, record: DecisionRecord
) -> int:
cursor = conn.execute(
"""
INSERT INTO decisions (
decision_type, proposal_id, timestamp,
inputs_json, outputs_json, action_taken, notes
) VALUES (?,?,?,?,?,?,?)
""",
(
record.decision_type,
_enc_uuid(record.proposal_id),
_enc_dt(record.timestamp),
record.inputs_json,
record.outputs_json,
record.action_taken,
record.notes,
),
)
return int(cursor.lastrowid or 0)
def list_decisions(
self,
conn: sqlite3.Connection,
*,
proposal_id: UUID | None = None,
limit: int = 100,
) -> list[DecisionRecord]:
if proposal_id is None:
rows = conn.execute(
"SELECT * FROM decisions ORDER BY timestamp DESC LIMIT ?",
(limit,),
).fetchall()
else:
rows = conn.execute(
"SELECT * FROM decisions WHERE proposal_id = ? "
"ORDER BY timestamp DESC LIMIT ?",
(_enc_uuid(proposal_id), limit),
).fetchall()
return [_row_to_decision(r) for r in rows]
# ------------------------------------------------------------------
# dvol_history
# ------------------------------------------------------------------
def record_dvol_snapshot(
self, conn: sqlite3.Connection, snapshot: DvolSnapshot
) -> None:
conn.execute(
"INSERT OR REPLACE INTO dvol_history(timestamp, asset, dvol, spot) "
"VALUES (?,?,?,?)",
(
_enc_dt(snapshot.timestamp),
snapshot.asset,
_enc_dec(snapshot.dvol),
_enc_dec(snapshot.spot),
),
)
# ------------------------------------------------------------------
# market_snapshots
# ------------------------------------------------------------------
def record_market_snapshot(
self, conn: sqlite3.Connection, snapshot: MarketSnapshotRecord
) -> None:
conn.execute(
"INSERT OR REPLACE INTO market_snapshots("
"timestamp, asset, spot, dvol, realized_vol_30d, iv_minus_rv, "
"funding_perp_annualized, funding_cross_annualized, "
"dealer_net_gamma, gamma_flip_level, oi_delta_pct_4h, "
"liquidation_long_risk, liquidation_short_risk, "
"macro_days_to_event, fetch_ok, fetch_errors_json) "
"VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)",
(
_enc_dt(snapshot.timestamp),
snapshot.asset,
_enc_dec(snapshot.spot),
_enc_dec(snapshot.dvol),
_enc_dec(snapshot.realized_vol_30d),
_enc_dec(snapshot.iv_minus_rv),
_enc_dec(snapshot.funding_perp_annualized),
_enc_dec(snapshot.funding_cross_annualized),
_enc_dec(snapshot.dealer_net_gamma),
_enc_dec(snapshot.gamma_flip_level),
_enc_dec(snapshot.oi_delta_pct_4h),
snapshot.liquidation_long_risk,
snapshot.liquidation_short_risk,
snapshot.macro_days_to_event,
1 if snapshot.fetch_ok else 0,
snapshot.fetch_errors_json,
),
)
def list_market_snapshots(
self,
conn: sqlite3.Connection,
*,
asset: str,
start: datetime | None = None,
end: datetime | None = None,
limit: int = 5000,
) -> list[MarketSnapshotRecord]:
clauses: list[str] = ["asset = ?"]
params: list[Any] = [asset]
if start is not None:
clauses.append("timestamp >= ?")
params.append(_enc_dt(start))
if end is not None:
clauses.append("timestamp <= ?")
params.append(_enc_dt(end))
params.append(int(limit))
rows = conn.execute(
f"SELECT * FROM market_snapshots WHERE {' AND '.join(clauses)} "
f"ORDER BY timestamp DESC LIMIT ?",
params,
).fetchall()
return [_row_to_market_snapshot(r) for r in rows]
def count_iv_rv_distinct_days(
self,
conn: sqlite3.Connection,
*,
asset: str,
max_days: int,
as_of: datetime | None = None,
) -> int:
"""Numero di giorni di calendario distinti coperti da IV-RV validi.
Esclude righe con ``fetch_ok=0`` o ``iv_minus_rv IS NULL``.
Usato dal caller del gate adattivo per decidere la finestra
(warmup hard / min_days / target_days).
Args:
as_of: Reference time for the rolling window. Defaults to
``datetime.now(UTC)``.
"""
if max_days <= 0:
raise ValueError(f"max_days must be positive, got {max_days}")
ref = as_of if as_of is not None else datetime.now(UTC)
if ref.tzinfo is None:
raise ValueError("as_of must be timezone-aware")
cutoff = ref - timedelta(days=max_days)
row = conn.execute(
"SELECT COUNT(DISTINCT substr(timestamp, 1, 10)) AS n "
"FROM market_snapshots "
"WHERE asset = ? "
" AND fetch_ok = 1 "
" AND iv_minus_rv IS NOT NULL "
" AND timestamp >= ?",
(asset, _enc_dt(cutoff)),
).fetchone()
return int(row["n"]) if row is not None else 0
def iv_rv_values_for_window(
self,
conn: sqlite3.Connection,
*,
asset: str,
window_days: int,
as_of: datetime | None = None,
) -> list[Decimal]:
"""Valori IV-RV ordinati ASC su ``[as_of - window_days, as_of]``.
Esclude righe con ``fetch_ok=0`` o ``iv_minus_rv IS NULL``.
Tutti i record validi della finestra concorrono come singoli
contributi alla statistica del percentile, indipendentemente
dalla cadenza con cui sono stati raccolti (tick live vs backfill
daily).
"""
if window_days <= 0:
raise ValueError(f"window_days must be positive, got {window_days}")
ref = as_of if as_of is not None else datetime.now(UTC)
if ref.tzinfo is None:
raise ValueError("as_of must be timezone-aware")
cutoff = ref - timedelta(days=window_days)
rows = conn.execute(
"SELECT iv_minus_rv FROM market_snapshots "
"WHERE asset = ? "
" AND fetch_ok = 1 "
" AND iv_minus_rv IS NOT NULL "
" AND timestamp >= ? "
"ORDER BY timestamp ASC",
(asset, _enc_dt(cutoff)),
).fetchall()
return [Decimal(str(r["iv_minus_rv"])) for r in rows]
def dvol_lookback(
self,
conn: sqlite3.Connection,
*,
asset: str,
reference: datetime,
tolerance_minutes: int = 15,
) -> Decimal | None:
"""DVOL al tick più vicino a `reference`, entro ±tolerance_minutes.
Ritorna ``None`` se non esiste un tick valido (``fetch_ok=1``,
``dvol IS NOT NULL``) entro la tolerance. Usato dal Vol-of-Vol
guard per stimare DVOL N ore fa.
"""
if reference.tzinfo is None:
raise ValueError("reference must be timezone-aware")
ref_lo = (reference - timedelta(minutes=tolerance_minutes)).astimezone(UTC)
ref_hi = (reference + timedelta(minutes=tolerance_minutes)).astimezone(UTC)
row = conn.execute(
"SELECT dvol, timestamp FROM market_snapshots "
"WHERE asset = ? "
" AND fetch_ok = 1 "
" AND dvol IS NOT NULL "
" AND timestamp >= ? "
" AND timestamp <= ? "
"ORDER BY ABS(julianday(timestamp) - julianday(?)) ASC LIMIT 1",
(
asset,
_enc_dt(ref_lo),
_enc_dt(ref_hi),
_enc_dt(reference),
),
).fetchone()
if row is None:
return None
return Decimal(str(row["dvol"]))
# ------------------------------------------------------------------
# option_chain_snapshots
# ------------------------------------------------------------------
def record_option_chain_snapshot(
self,
conn: sqlite3.Connection,
quotes: list[OptionChainQuoteRecord],
) -> int:
"""Bulk-insert dei quote di un singolo tick. Tutti i quote
condividono lo stesso ``timestamp``. Idempotente per
(timestamp, instrument_name)."""
if not quotes:
return 0
rows = [
(
_enc_dt(q.timestamp),
q.asset,
q.instrument_name,
_enc_dec(q.strike),
_enc_dt(q.expiry),
q.option_type,
_enc_dec(q.bid),
_enc_dec(q.ask),
_enc_dec(q.mid),
_enc_dec(q.iv),
_enc_dec(q.delta),
_enc_dec(q.gamma),
_enc_dec(q.theta),
_enc_dec(q.vega),
q.open_interest,
q.volume_24h,
q.book_depth_top3,
)
for q in quotes
]
conn.executemany(
"INSERT OR REPLACE INTO option_chain_snapshots("
"timestamp, asset, instrument_name, strike, expiry, option_type, "
"bid, ask, mid, iv, delta, gamma, theta, vega, "
"open_interest, volume_24h, book_depth_top3) "
"VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)",
rows,
)
return len(rows)
def list_option_chain_snapshots(
self,
conn: sqlite3.Connection,
*,
asset: str,
start: datetime | None = None,
end: datetime | None = None,
expiry_from: datetime | None = None,
expiry_to: datetime | None = None,
limit: int = 50000,
) -> list[OptionChainQuoteRecord]:
clauses: list[str] = ["asset = ?"]
params: list[Any] = [asset]
if start is not None:
clauses.append("timestamp >= ?")
params.append(_enc_dt(start))
if end is not None:
clauses.append("timestamp <= ?")
params.append(_enc_dt(end))
if expiry_from is not None:
clauses.append("expiry >= ?")
params.append(_enc_dt(expiry_from))
if expiry_to is not None:
clauses.append("expiry <= ?")
params.append(_enc_dt(expiry_to))
params.append(int(limit))
rows = conn.execute(
f"SELECT * FROM option_chain_snapshots "
f"WHERE {' AND '.join(clauses)} "
f"ORDER BY timestamp DESC, instrument_name ASC LIMIT ?",
params,
).fetchall()
return [_row_to_option_chain_quote(r) for r in rows]
def latest_option_chain_timestamp(
self,
conn: sqlite3.Connection,
*,
asset: str,
) -> datetime | None:
"""Timestamp dell'ultimo snapshot raccolto per ``asset``,
utile per stimare la freschezza del dato dalla GUI."""
row = conn.execute(
"SELECT timestamp FROM option_chain_snapshots "
"WHERE asset = ? ORDER BY timestamp DESC LIMIT 1",
(asset,),
).fetchone()
if row is None:
return None
return _dec_dt(row["timestamp"])
# ------------------------------------------------------------------
# manual_actions
# ------------------------------------------------------------------
def enqueue_manual_action(
self, conn: sqlite3.Connection, action: ManualAction
) -> int:
cursor = conn.execute(
"INSERT INTO manual_actions(kind, proposal_id, payload_json, created_at) "
"VALUES (?,?,?,?)",
(
action.kind,
_enc_uuid(action.proposal_id),
action.payload_json,
_enc_dt(action.created_at),
),
)
return int(cursor.lastrowid or 0)
def next_unconsumed_action(
self, conn: sqlite3.Connection
) -> ManualAction | None:
row = conn.execute(
"SELECT * FROM manual_actions WHERE consumed_at IS NULL "
"ORDER BY created_at ASC LIMIT 1"
).fetchone()
return None if row is None else _row_to_manual(row)
def mark_action_consumed(
self,
conn: sqlite3.Connection,
action_id: int,
*,
consumed_by: str,
result: str,
now: datetime,
) -> None:
conn.execute(
"UPDATE manual_actions SET consumed_at = ?, consumed_by = ?, "
"result = ? WHERE id = ?",
(_enc_dt(now), consumed_by, result, action_id),
)
# ------------------------------------------------------------------
# system_state
# ------------------------------------------------------------------
def init_system_state(
self, conn: sqlite3.Connection, *, config_version: str, now: datetime
) -> None:
"""Insert the singleton row if it does not already exist."""
existing = conn.execute(
"SELECT 1 FROM system_state WHERE id = 1"
).fetchone()
if existing is not None:
return
conn.execute(
"INSERT INTO system_state(id, kill_switch, last_health_check, "
"config_version, started_at) VALUES (1, 0, ?, ?, ?)",
(_enc_dt(now), config_version, _enc_dt(now)),
)
def get_system_state(
self, conn: sqlite3.Connection
) -> SystemStateRecord | None:
row = conn.execute("SELECT * FROM system_state WHERE id = 1").fetchone()
if row is None:
return None
keys = row.keys()
return SystemStateRecord(
id=int(row["id"]),
kill_switch=int(row["kill_switch"]),
kill_reason=row["kill_reason"],
kill_at=_dec_dt(row["kill_at"]),
last_health_check=_dec_dt_required(row["last_health_check"]),
last_kelly_calib=_dec_dt(row["last_kelly_calib"]),
config_version=row["config_version"],
started_at=_dec_dt_required(row["started_at"]),
last_audit_hash=(
row["last_audit_hash"] if "last_audit_hash" in keys else None
),
auto_pause_until=(
_dec_dt(row["auto_pause_until"])
if "auto_pause_until" in keys
else None
),
auto_pause_reason=(
row["auto_pause_reason"]
if "auto_pause_reason" in keys
else None
),
)
def set_last_audit_hash(
self, conn: sqlite3.Connection, *, hex_hash: str
) -> None:
"""Store the most recent audit chain hash. Called by AuditLog after append."""
conn.execute(
"UPDATE system_state SET last_audit_hash = ? WHERE id = 1",
(hex_hash,),
)
def set_kill_switch(
self,
conn: sqlite3.Connection,
*,
armed: bool,
reason: str | None,
now: datetime,
) -> None:
conn.execute(
"UPDATE system_state SET kill_switch = ?, kill_reason = ?, "
"kill_at = ?, last_health_check = ? WHERE id = 1",
(
1 if armed else 0,
reason,
_enc_dt(now) if armed else None,
_enc_dt(now),
),
)
def touch_health_check(
self, conn: sqlite3.Connection, *, now: datetime
) -> None:
conn.execute(
"UPDATE system_state SET last_health_check = ? WHERE id = 1",
(_enc_dt(now),),
)
def set_auto_pause(
self,
conn: sqlite3.Connection,
*,
until: datetime | None,
reason: str | None,
) -> None:
"""Imposta o azzera la pausa automatica (§7-bis F).
``until = None`` annulla la pausa (l'engine torna attivo).
Il setter è idempotente: chiamarlo con un until già nel passato
è equivalente a clear.
"""
conn.execute(
"UPDATE system_state SET auto_pause_until = ?, "
"auto_pause_reason = ? WHERE id = 1",
(_enc_dt(until) if until is not None else None, reason),
)
def recent_closed_position_pnls_usd(
self, conn: sqlite3.Connection, *, limit: int
) -> list[Decimal]:
"""Ritorna la lista dei pnl_usd delle ultime ``limit`` posizioni chiuse,
ordinate dalla più recente alla più vecchia. Posizioni con
``pnl_usd`` ``NULL`` (es. chiuse di emergenza senza P/L noto)
sono saltate. Usato dal circuit breaker §7-bis F.
"""
if limit <= 0:
return []
rows = conn.execute(
"SELECT pnl_usd FROM positions "
"WHERE closed_at IS NOT NULL AND pnl_usd IS NOT NULL "
"ORDER BY closed_at DESC LIMIT ?",
(limit,),
).fetchall()
return [Decimal(row["pnl_usd"]) for row in rows]
# ---------------------------------------------------------------------------
# Row → model converters
# ---------------------------------------------------------------------------
def _dec_dt_required(value: Any) -> datetime:
out = _dec_dt(value)
if out is None:
raise ValueError("expected non-null datetime in row")
return out
def _row_to_position(row: sqlite3.Row) -> PositionRecord:
proposal_id = _dec_uuid(row["proposal_id"])
if proposal_id is None:
raise ValueError("positions.proposal_id was NULL")
return PositionRecord(
proposal_id=proposal_id,
spread_type=row["spread_type"],
asset=row["asset"],
expiry=_dec_dt_required(row["expiry"]),
short_strike=_dec_dec_required(row["short_strike"]),
long_strike=_dec_dec_required(row["long_strike"]),
short_instrument=row["short_instrument"],
long_instrument=row["long_instrument"],
n_contracts=int(row["n_contracts"]),
spread_width_usd=_dec_dec_required(row["spread_width_usd"]),
spread_width_pct=_dec_dec_required(row["spread_width_pct"]),
credit_eth=_dec_dec_required(row["credit_eth"]),
credit_usd=_dec_dec_required(row["credit_usd"]),
max_loss_usd=_dec_dec_required(row["max_loss_usd"]),
spot_at_entry=_dec_dec_required(row["spot_at_entry"]),
dvol_at_entry=_dec_dec_required(row["dvol_at_entry"]),
delta_at_entry=_dec_dec_required(row["delta_at_entry"]),
eth_price_at_entry=_dec_dec_required(row["eth_price_at_entry"]),
proposed_at=_dec_dt_required(row["proposed_at"]),
opened_at=_dec_dt(row["opened_at"]),
closed_at=_dec_dt(row["closed_at"]),
close_reason=row["close_reason"],
debit_paid_eth=_dec_dec(row["debit_paid_eth"]),
pnl_eth=_dec_dec(row["pnl_eth"]),
pnl_usd=_dec_dec(row["pnl_usd"]),
status=row["status"],
created_at=_dec_dt_required(row["created_at"]),
updated_at=_dec_dt_required(row["updated_at"]),
)
def _row_to_instruction(row: sqlite3.Row) -> InstructionRecord:
instruction_id = _dec_uuid(row["instruction_id"])
proposal_id = _dec_uuid(row["proposal_id"])
if instruction_id is None or proposal_id is None:
raise ValueError("instructions row missing required UUID")
return InstructionRecord(
instruction_id=instruction_id,
proposal_id=proposal_id,
kind=row["kind"],
payload_json=row["payload_json"],
sent_at=_dec_dt_required(row["sent_at"]),
acknowledged_at=_dec_dt(row["acknowledged_at"]),
filled_at=_dec_dt(row["filled_at"]),
cancelled_at=_dec_dt(row["cancelled_at"]),
actual_fill_eth=_dec_dec(row["actual_fill_eth"]),
actual_fees_eth=_dec_dec(row["actual_fees_eth"]),
)
def _row_to_decision(row: sqlite3.Row) -> DecisionRecord:
return DecisionRecord(
id=int(row["id"]),
decision_type=row["decision_type"],
proposal_id=_dec_uuid(row["proposal_id"]),
timestamp=_dec_dt_required(row["timestamp"]),
inputs_json=row["inputs_json"],
outputs_json=row["outputs_json"],
action_taken=row["action_taken"],
notes=row["notes"],
)
def _row_to_manual(row: sqlite3.Row) -> ManualAction:
return ManualAction(
id=int(row["id"]),
kind=row["kind"],
proposal_id=_dec_uuid(row["proposal_id"]),
payload_json=row["payload_json"],
created_at=_dec_dt_required(row["created_at"]),
consumed_at=_dec_dt(row["consumed_at"]),
consumed_by=row["consumed_by"],
result=row["result"],
)
def _row_to_market_snapshot(row: sqlite3.Row) -> MarketSnapshotRecord:
return MarketSnapshotRecord(
timestamp=_dec_dt_required(row["timestamp"]),
asset=row["asset"],
spot=_dec_dec(row["spot"]),
dvol=_dec_dec(row["dvol"]),
realized_vol_30d=_dec_dec(row["realized_vol_30d"]),
iv_minus_rv=_dec_dec(row["iv_minus_rv"]),
funding_perp_annualized=_dec_dec(row["funding_perp_annualized"]),
funding_cross_annualized=_dec_dec(row["funding_cross_annualized"]),
dealer_net_gamma=_dec_dec(row["dealer_net_gamma"]),
gamma_flip_level=_dec_dec(row["gamma_flip_level"]),
oi_delta_pct_4h=_dec_dec(row["oi_delta_pct_4h"]),
liquidation_long_risk=row["liquidation_long_risk"],
liquidation_short_risk=row["liquidation_short_risk"],
macro_days_to_event=(
int(row["macro_days_to_event"])
if row["macro_days_to_event"] is not None
else None
),
fetch_ok=bool(int(row["fetch_ok"])),
fetch_errors_json=row["fetch_errors_json"],
)
def _row_to_option_chain_quote(row: sqlite3.Row) -> OptionChainQuoteRecord:
return OptionChainQuoteRecord(
timestamp=_dec_dt_required(row["timestamp"]),
asset=row["asset"],
instrument_name=row["instrument_name"],
strike=_dec_dec_required(row["strike"]),
expiry=_dec_dt_required(row["expiry"]),
option_type=row["option_type"],
bid=_dec_dec(row["bid"]),
ask=_dec_dec(row["ask"]),
mid=_dec_dec(row["mid"]),
iv=_dec_dec(row["iv"]),
delta=_dec_dec(row["delta"]),
gamma=_dec_dec(row["gamma"]),
theta=_dec_dec(row["theta"]),
vega=_dec_dec(row["vega"]),
open_interest=(
int(row["open_interest"])
if row["open_interest"] is not None
else None
),
volume_24h=(
int(row["volume_24h"]) if row["volume_24h"] is not None else None
),
book_depth_top3=(
int(row["book_depth_top3"])
if row["book_depth_top3"] is not None
else None
),
)
def _dec_dec_required(value: Any) -> Decimal:
out = _dec_dec(value)
if out is None:
raise ValueError("expected non-null Decimal in row")
return out
Reference in New Issue View Git Blame Copy Permalink