trex/trex-core
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(reconcile): safely seed canonical cost basis

Browse Source
This commit is contained in:
Hermes Agent
2026-07-17 09:57:08 +09:00
parent 18bd8d9f0b
commit 81a853a314
25 changed files with 1689 additions and 66 deletions
Download Patch File Download Diff File Expand all files Collapse all files
trex_core/application/reconciliation.py
+175 -15
View File
@@ -23,7 +23,9 @@ BAT-051 / REQ-106: verify_slot_reconcile — operator CLI verify path (C-only, r
from __future__ import annotations
import logging
from typing import Optional
import math
import re
from typing import Optional, cast
from trex_core.application.alarm_emitter import AlarmEmitter
from trex_core.application.broker_port import BrokerPort
@@ -32,6 +34,68 @@ from trex_core.domain.slot.state import SlotState
_log = logging.getLogger(__name__)
_COST_BASIS_REASON_CODES = frozenset({
"missing",
"invalid",
"transport",
"invalid_json",
"invalid_payload",
"not_configured",
"diagnostic_exception",
"diagnostic_source_invalid",
})
_COST_BASIS_LEGACY_SOURCE_ALIASES = frozenset({"avg_cost", "wac", "cost_basis"})
def _sanitize_cost_basis_reason(reason: object) -> Optional[str]:
"""Keep public diagnostics in the bounded REQ-123 reason taxonomy."""
if not isinstance(reason, str):
return None
if reason in _COST_BASIS_REASON_CODES:
return reason
if re.fullmatch(r"http_status_\d{3}", reason):
return reason
return None
def _normalize_cost_basis_diagnostics(
value: object,
source: object,
reason: object,
) -> tuple[Optional[float], str, Optional[str]]:
"""Normalize diagnostic-port output before it reaches reconciliation results."""
safe_reason = _sanitize_cost_basis_reason(reason)
if source == "canonical":
normalized_source = "canonical"
elif source == "legacy" or source in _COST_BASIS_LEGACY_SOURCE_ALIASES:
normalized_source = "legacy"
elif source == "none":
normalized_source = "none"
else:
return None, "none", safe_reason or "diagnostic_source_invalid"
if value is None:
return None, normalized_source, safe_reason
if isinstance(value, bool):
return None, "none", safe_reason or "invalid_payload"
try:
normalized_value = float(cast(str | int | float, value))
except (TypeError, ValueError):
return None, "none", safe_reason or "invalid_payload"
if not math.isfinite(normalized_value) or normalized_value <= 0:
return None, "none", safe_reason or "invalid_payload"
if normalized_source == "none":
return None, "none", safe_reason or "invalid_payload"
return normalized_value, normalized_source, safe_reason
def _normalize_legacy_cost_basis(value: object) -> tuple[Optional[float], str, Optional[str]]:
"""Make the diagnostic-free compatibility seam obey safe WAC semantics."""
normalized_value, _, _ = _normalize_cost_basis_diagnostics(value, "legacy", "invalid")
if normalized_value is None:
return None, "none", "invalid"
return normalized_value, "legacy", None
def reconcile_core_holdings(
*,
@@ -308,25 +372,114 @@ def sync_slot_to_broker(
# Try to seed avg_price from portfolio-hub cost-basis port (FR-7, graceful degrade).
avg_price = slot.avg_price
seeded_avg_price = False
cost_basis_source = "none"
cost_basis_reason: Optional[str] = None
def _set_cost_basis_unavailable_message() -> tuple[str, Optional[str]]:
if pid:
return "cost_basis_unavailable", f"inactive slot {slot_id} portfolio_id={pid}"
return "cost_basis_unavailable", "inactive slot without portfolio_id"
if cost_basis_port is not None and pid:
try:
wac = cost_basis_port.get_cost_basis(pid)
if wac is not None:
avg_price = wac
seeded_avg_price = True
except Exception: # noqa: BLE001 — graceful degrade
_log.warning(
"sync: cost_basis_port.get_cost_basis failed for portfolio_id=%s slot=%s "
"— skipping avg_price seed (graceful degrade)",
pid, slot_id,
)
elif cost_basis_port is None and pid:
diagnostics = getattr(cost_basis_port, "get_cost_basis_with_diagnostics", None)
if callable(diagnostics):
try:
result = diagnostics(pid)
if isinstance(result, dict):
raw_value = result.get("value")
raw_source = result.get("source")
raw_reason = result.get("reason")
elif isinstance(result, tuple) and len(result) >= 1:
raw_value = result[0]
raw_source = result[1] if len(result) >= 2 else None
raw_reason = result[2] if len(result) >= 3 else None
else:
raise TypeError("invalid_cost_basis_diagnostic_format")
cost_basis_value, cost_basis_source, cost_basis_reason = (
_normalize_cost_basis_diagnostics(raw_value, raw_source, raw_reason)
)
if cost_basis_value is not None:
avg_price = cost_basis_value
seeded_avg_price = True
cost_basis_reason = None
except Exception: # noqa: BLE001 — graceful degrade
cost_basis_source = "none"
cost_basis_reason = "diagnostic_exception"
_log.warning(
"sync: cost_basis_port.get_cost_basis_with_diagnostics failed for "
"portfolio_id=%s slot=%s — skipping avg_price seed (graceful degrade)",
pid, slot_id,
)
else:
# Backward compatibility path if diagnostic method is not available.
try:
legacy_value, legacy_source, legacy_reason = _normalize_legacy_cost_basis(
cost_basis_port.get_cost_basis(pid)
)
cost_basis_source = legacy_source
cost_basis_reason = legacy_reason
if legacy_value is not None:
avg_price = legacy_value
seeded_avg_price = True
except Exception: # noqa: BLE001 — graceful degrade
cost_basis_source = "none"
cost_basis_reason = "diagnostic_exception"
_log.warning(
"sync: cost_basis_port.get_cost_basis failed for portfolio_id=%s slot=%s "
"— skipping avg_price seed (graceful degrade)",
pid, slot_id,
)
elif pid:
cost_basis_reason = "not_configured"
_log.warning(
"sync: no cost_basis_port configured for slot=%s portfolio_id=%s "
"— skipping avg_price seed (graceful degrade)",
slot_id, pid,
)
inactive_no_cost_basis = slot.status is SlotState.INACTIVE and avg_price is None and broker_qty > 0
fail_closed_cost_basis = inactive_no_cost_basis and not dry_run
if inactive_no_cost_basis and dry_run:
return {
"slot_id": slot_id,
"symbol": sym,
"portfolio_id": pid,
"before": before,
"after": before,
"warning_cleared": False,
"dry_run": True,
"skipped": True,
"skip_reason": _set_cost_basis_unavailable_message()[0],
"seeded_avg_price": False,
"cost_basis_source": cost_basis_source,
"cost_basis_reason": cost_basis_reason,
}
if fail_closed_cost_basis:
_log.error(
"sync fail-closed: INACTIVE onboarding requires avg_price seed (portfolio_id=%s slot=%s) "
"— refuse live write",
pid,
slot_id,
)
return {
"slot_id": slot_id,
"symbol": sym,
"portfolio_id": pid,
"before": before,
"after": before,
"warning_cleared": False,
"dry_run": False,
"skipped": True,
"skip_reason": _set_cost_basis_unavailable_message()[0],
"seeded_avg_price": False,
"cost_basis_source": cost_basis_source,
"cost_basis_reason": cost_basis_reason,
"error": "cost_basis_unavailable",
}
if dry_run:
# Compute what *would* happen without writing anything.
new_status = SlotState.HOLDING if slot.status is SlotState.INACTIVE else slot.status
@@ -343,6 +496,8 @@ def sync_slot_to_broker(
"dry_run": True,
"skipped": False,
"seeded_avg_price": seeded_avg_price,
"cost_basis_source": cost_basis_source,
"cost_basis_reason": cost_basis_reason,
}
# --- Live write ---
@@ -374,11 +529,15 @@ def sync_slot_to_broker(
_log.exception("sync_slot_to_broker: write failed for slot=%s", slot_id)
return {
"slot_id": slot_id, "symbol": sym, "portfolio_id": pid,
"before": before, "after": before,
"before": before,
"after": before,
"warning_cleared": False,
"dry_run": False,
"skipped": False,
"error": "write_failed",
"seeded_avg_price": seeded_avg_price,
"cost_basis_source": cost_basis_source,
"cost_basis_reason": cost_basis_reason,
}
# Refresh to confirm
@@ -409,9 +568,10 @@ def sync_slot_to_broker(
"dry_run": False,
"skipped": False,
"seeded_avg_price": seeded_avg_price,
"cost_basis_source": cost_basis_source,
"cost_basis_reason": cost_basis_reason,
}
# ---------------------------------------------------------------------------
# BAT-051 / REQ-106: operator verify command — C-case re-check (read-centric)
# ---------------------------------------------------------------------------