Adriano Dal Pastro
4d7b02b6cb
Il backup faceva uno snapshot full del DB ogni ora: ~16GB accumulati per ~98MB di dato unico (218k righe option_chain ricopiate ~337 volte), con retention 30g. Porta il cron a giornaliero (0 0 * * *): -24x lo spazio a parita' di retention. Rilevante prima di accendere il collettore research (che 2-4x la crescita del DB e quindi dei backup). Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
503 lines
19 KiB
Python
503 lines
19 KiB
Python
"""Façade that ties the runtime modules into a runnable engine.
|
|
|
|
The :class:`Orchestrator` is the single entry point for the CLI: it
|
|
holds the :class:`RuntimeContext`, the :class:`HealthCheck` state, and
|
|
the boot procedure (recover + boot environment check + scheduler
|
|
arming). Every concrete cycle is delegated to its own module so each
|
|
piece stays independently testable.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import logging
|
|
from collections.abc import Awaitable, Callable
|
|
from dataclasses import dataclass
|
|
from datetime import UTC, datetime
|
|
from decimal import Decimal
|
|
from pathlib import Path
|
|
from typing import Literal
|
|
from uuid import uuid4
|
|
|
|
import structlog
|
|
from apscheduler.schedulers.asyncio import AsyncIOScheduler
|
|
|
|
from cerbero_bite.config.mcp_endpoints import McpEndpoints
|
|
from cerbero_bite.config.runtime_flags import RuntimeFlags
|
|
from cerbero_bite.config.schema import StrategyConfig
|
|
from cerbero_bite.runtime.dependencies import RuntimeContext, build_runtime
|
|
from cerbero_bite.runtime.entry_cycle import EntryCycleResult, run_entry_cycle
|
|
from cerbero_bite.runtime.health_check import HealthCheck, HealthCheckResult
|
|
from cerbero_bite.runtime.lockfile import EngineLock
|
|
from cerbero_bite.runtime.manual_actions_consumer import consume_manual_actions
|
|
from cerbero_bite.runtime.market_snapshot_cycle import (
|
|
DEFAULT_ASSETS,
|
|
collect_market_snapshot,
|
|
)
|
|
from cerbero_bite.runtime.option_chain_snapshot_cycle import (
|
|
collect_option_chain_snapshot,
|
|
)
|
|
from cerbero_bite.runtime.option_chain_research_cycle import (
|
|
collect_option_chain_research,
|
|
)
|
|
from cerbero_bite.runtime.monitor_cycle import MonitorCycleResult, run_monitor_cycle
|
|
from cerbero_bite.runtime.recovery import recover_state
|
|
from cerbero_bite.runtime.scheduler import JobSpec, build_scheduler
|
|
from cerbero_bite.safety.audit_log import tail_continues_from
|
|
from cerbero_bite.state import connect as connect_state
|
|
from cerbero_bite.state import transaction
|
|
|
|
__all__ = ["Orchestrator"]
|
|
|
|
|
|
_log = logging.getLogger("cerbero_bite.runtime.orchestrator")
|
|
|
|
Environment = Literal["testnet", "mainnet"]
|
|
|
|
# Default cron schedule (matches docs/06-operational-flow.md table).
|
|
_CRON_ENTRY = "0 */2 * * *" # crypto 24/7: valutazione ogni 2h; i gate decidono se entrare
|
|
_CRON_MONITOR = "0 * * * *" # stop/take-profit check ogni ora
|
|
_CRON_HEALTH = "*/5 * * * *"
|
|
_CRON_BACKUP = "0 0 * * *" # giornaliero (00:00 UTC): lo snapshot full orario gonfiava i backup (16GB per ~98MB di dato unico)
|
|
_CRON_MANUAL_ACTIONS = "*/1 * * * *"
|
|
_CRON_MARKET_SNAPSHOT = "*/15 * * * *"
|
|
_CRON_OPTION_CHAIN_SNAPSHOT = "*/15 * * * *" # crypto è 24/7: cadenza continua allineata a market_snapshot
|
|
_BACKUP_RETENTION_DAYS = 30
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class _BootResult:
|
|
environment: Environment
|
|
health: HealthCheckResult
|
|
|
|
|
|
class Orchestrator:
|
|
"""Engine façade — boot, scheduler, manual cycle invocation."""
|
|
|
|
def __init__(
|
|
self,
|
|
ctx: RuntimeContext,
|
|
*,
|
|
expected_environment: Environment,
|
|
eur_to_usd: Decimal,
|
|
flags: RuntimeFlags | None = None,
|
|
) -> None:
|
|
self._ctx = ctx
|
|
self._expected_env = expected_environment
|
|
self._eur_to_usd = eur_to_usd
|
|
self._flags = flags or RuntimeFlags()
|
|
self._health = HealthCheck(ctx, expected_environment=expected_environment)
|
|
self._scheduler: AsyncIOScheduler | None = None
|
|
|
|
@property
|
|
def context(self) -> RuntimeContext:
|
|
return self._ctx
|
|
|
|
@property
|
|
def expected_environment(self) -> Environment:
|
|
return self._expected_env
|
|
|
|
@property
|
|
def flags(self) -> RuntimeFlags:
|
|
return self._flags
|
|
|
|
# ------------------------------------------------------------------
|
|
# Boot
|
|
# ------------------------------------------------------------------
|
|
|
|
async def boot(self) -> _BootResult:
|
|
"""Reconcile state, verify environment, run a first health probe."""
|
|
when = self._ctx.clock()
|
|
await self._verify_audit_anchor(now=when)
|
|
await recover_state(self._ctx, now=when)
|
|
|
|
info = await self._ctx.deribit.environment_info()
|
|
if info.environment != self._expected_env:
|
|
await self._ctx.alert_manager.critical(
|
|
source="orchestrator.boot",
|
|
message=(
|
|
f"Deribit environment mismatch at boot: expected "
|
|
f"{self._expected_env}, got {info.environment}"
|
|
),
|
|
component="runtime.orchestrator",
|
|
)
|
|
|
|
health = await self._health.run(now=when)
|
|
self._ctx.audit_log.append(
|
|
event="ENGINE_START",
|
|
payload={
|
|
"environment": info.environment,
|
|
"health": health.state,
|
|
"config_version": self._ctx.cfg.config_version,
|
|
"data_analysis_enabled": self._flags.data_analysis_enabled,
|
|
"strategy_enabled": self._flags.strategy_enabled,
|
|
},
|
|
now=when,
|
|
)
|
|
_log.info(
|
|
"engine started: env=%s health=%s data_analysis=%s strategy=%s",
|
|
info.environment,
|
|
health.state,
|
|
self._flags.data_analysis_enabled,
|
|
self._flags.strategy_enabled,
|
|
)
|
|
return _BootResult(environment=info.environment, health=health)
|
|
|
|
# ------------------------------------------------------------------
|
|
# Cycle invocations (used by scheduler jobs and CLI dry-run)
|
|
# ------------------------------------------------------------------
|
|
|
|
async def _verify_audit_anchor(self, *, now: datetime) -> None: # noqa: ARG002
|
|
"""Compare the audit log tail with the SQLite anchor.
|
|
|
|
``now`` is accepted for symmetry with the other ``boot``
|
|
helpers but unused: the comparison is purely between the
|
|
in-memory tail hash and the value persisted on the previous
|
|
run.
|
|
"""
|
|
conn = connect_state(self._ctx.db_path)
|
|
try:
|
|
state = self._ctx.repository.get_system_state(conn)
|
|
finally:
|
|
conn.close()
|
|
if state is None or state.last_audit_hash is None:
|
|
return # first boot, nothing to compare against
|
|
actual_tail = self._ctx.audit_log.last_hash
|
|
if actual_tail == state.last_audit_hash:
|
|
return
|
|
# The anchor is persisted best-effort (see build_runtime): under
|
|
# SQLite write contention the mirror can fall behind the log while
|
|
# the log itself keeps growing forward, intact. Treat that benign
|
|
# lag — anchor is a valid ancestor of the current tail — as a
|
|
# re-sync, not tampering. Only a missing anchor or a broken
|
|
# post-anchor chain (truncation/tampering) arms the kill switch.
|
|
if tail_continues_from(self._ctx.audit_log.path, state.last_audit_hash):
|
|
conn = connect_state(self._ctx.db_path)
|
|
try:
|
|
with transaction(conn):
|
|
self._ctx.repository.set_last_audit_hash(
|
|
conn, hex_hash=actual_tail
|
|
)
|
|
finally:
|
|
conn.close()
|
|
_log.warning(
|
|
"audit anchor lagged behind tail; re-synced "
|
|
"(anchor=%s…, tail=%s…)",
|
|
state.last_audit_hash[:12],
|
|
actual_tail[:12],
|
|
)
|
|
return
|
|
await self._ctx.alert_manager.critical(
|
|
source="orchestrator.boot",
|
|
message=(
|
|
f"audit log anchor mismatch: anchor="
|
|
f"{state.last_audit_hash[:12]}…, file tail="
|
|
f"{actual_tail[:12]}… — possible tampering or truncation"
|
|
),
|
|
component="safety.audit_log",
|
|
)
|
|
|
|
async def run_entry(
|
|
self, *, now: datetime | None = None
|
|
) -> EntryCycleResult:
|
|
cycle_id = str(uuid4())
|
|
token = structlog.contextvars.bind_contextvars(
|
|
cycle="entry", cycle_id=cycle_id
|
|
)
|
|
try:
|
|
return await run_entry_cycle(
|
|
self._ctx, eur_to_usd_rate=self._eur_to_usd, now=now
|
|
)
|
|
finally:
|
|
structlog.contextvars.reset_contextvars(**token)
|
|
|
|
async def run_monitor(
|
|
self, *, now: datetime | None = None
|
|
) -> MonitorCycleResult:
|
|
cycle_id = str(uuid4())
|
|
token = structlog.contextvars.bind_contextvars(
|
|
cycle="monitor", cycle_id=cycle_id
|
|
)
|
|
try:
|
|
return await run_monitor_cycle(self._ctx, now=now)
|
|
finally:
|
|
structlog.contextvars.reset_contextvars(**token)
|
|
|
|
async def run_health(
|
|
self, *, now: datetime | None = None
|
|
) -> HealthCheckResult:
|
|
cycle_id = str(uuid4())
|
|
token = structlog.contextvars.bind_contextvars(
|
|
cycle="health", cycle_id=cycle_id
|
|
)
|
|
try:
|
|
return await self._health.run(now=now)
|
|
finally:
|
|
structlog.contextvars.reset_contextvars(**token)
|
|
|
|
# ------------------------------------------------------------------
|
|
# Scheduler lifecycle
|
|
# ------------------------------------------------------------------
|
|
|
|
def install_scheduler(
|
|
self,
|
|
*,
|
|
entry_cron: str = _CRON_ENTRY,
|
|
monitor_cron: str = _CRON_MONITOR,
|
|
health_cron: str = _CRON_HEALTH,
|
|
backup_cron: str = _CRON_BACKUP,
|
|
manual_actions_cron: str = _CRON_MANUAL_ACTIONS,
|
|
market_snapshot_cron: str = _CRON_MARKET_SNAPSHOT,
|
|
market_snapshot_assets: tuple[str, ...] = DEFAULT_ASSETS,
|
|
option_chain_cron: str = _CRON_OPTION_CHAIN_SNAPSHOT,
|
|
option_chain_assets: tuple[str, ...] = ("ETH", "BTC"),
|
|
backup_dir: Path | None = None,
|
|
backup_retention_days: int = _BACKUP_RETENTION_DAYS,
|
|
) -> AsyncIOScheduler:
|
|
"""Build the scheduler with the canonical job set, ready to start."""
|
|
|
|
async def _safe(name: str, coro_factory: Callable[[], Awaitable[object]]) -> None:
|
|
try:
|
|
await coro_factory()
|
|
except Exception as exc: # never let a tick kill the scheduler
|
|
_log.exception("scheduler tick %s raised", name)
|
|
await self._ctx.alert_manager.critical(
|
|
source=f"scheduler.{name}",
|
|
message=f"{type(exc).__name__}: {exc}",
|
|
component=f"runtime.{name}",
|
|
)
|
|
|
|
async def _entry() -> None:
|
|
await _safe("entry", self.run_entry)
|
|
|
|
async def _monitor() -> None:
|
|
await _safe("monitor", self.run_monitor)
|
|
|
|
async def _health() -> None:
|
|
await _safe("health", self.run_health)
|
|
|
|
backups_target = backup_dir or self._ctx.db_path.parent / "backups"
|
|
|
|
async def _backup() -> None:
|
|
async def _do() -> None:
|
|
await asyncio.to_thread(
|
|
_run_backup,
|
|
db_path=self._ctx.db_path,
|
|
backup_dir=backups_target,
|
|
retention_days=backup_retention_days,
|
|
)
|
|
|
|
await _safe("backup", _do)
|
|
|
|
async def _run_market_snapshot_via_action() -> None:
|
|
await collect_market_snapshot(
|
|
self._ctx, assets=market_snapshot_assets
|
|
)
|
|
|
|
async def _manual_actions() -> None:
|
|
async def _do() -> None:
|
|
await consume_manual_actions(
|
|
self._ctx,
|
|
cycle_runners={
|
|
"entry": self.run_entry,
|
|
"monitor": self.run_monitor,
|
|
"health": self.run_health,
|
|
"market_snapshot": _run_market_snapshot_via_action,
|
|
},
|
|
)
|
|
|
|
await _safe("manual_actions", _do)
|
|
|
|
async def _market_snapshot() -> None:
|
|
async def _do() -> None:
|
|
await collect_market_snapshot(
|
|
self._ctx, assets=market_snapshot_assets
|
|
)
|
|
|
|
await _safe("market_snapshot", _do)
|
|
|
|
async def _option_chain_snapshot() -> None:
|
|
async def _do() -> None:
|
|
for asset in option_chain_assets:
|
|
await collect_option_chain_snapshot(self._ctx, asset=asset)
|
|
|
|
await _safe("option_chain_snapshot", _do)
|
|
|
|
async def _option_chain_research() -> None:
|
|
async def _do() -> None:
|
|
for asset in self._ctx.cfg.research_collector.assets:
|
|
await collect_option_chain_research(self._ctx, asset=asset)
|
|
|
|
await _safe("option_chain_research", _do)
|
|
|
|
jobs: list[JobSpec] = [
|
|
JobSpec(name="health", cron=health_cron, coro_factory=_health),
|
|
JobSpec(name="backup", cron=backup_cron, coro_factory=_backup),
|
|
JobSpec(
|
|
name="manual_actions",
|
|
cron=manual_actions_cron,
|
|
coro_factory=_manual_actions,
|
|
),
|
|
]
|
|
if self._flags.strategy_enabled:
|
|
jobs.append(JobSpec(name="entry", cron=entry_cron, coro_factory=_entry))
|
|
jobs.append(
|
|
JobSpec(name="monitor", cron=monitor_cron, coro_factory=_monitor)
|
|
)
|
|
else:
|
|
_log.warning(
|
|
"strategy disabled (CERBERO_BITE_ENABLE_STRATEGY=false): "
|
|
"entry and monitor cycles are NOT scheduled"
|
|
)
|
|
if self._flags.data_analysis_enabled:
|
|
jobs.append(
|
|
JobSpec(
|
|
name="market_snapshot",
|
|
cron=market_snapshot_cron,
|
|
coro_factory=_market_snapshot,
|
|
)
|
|
)
|
|
jobs.append(
|
|
JobSpec(
|
|
name="option_chain_snapshot",
|
|
cron=option_chain_cron,
|
|
coro_factory=_option_chain_snapshot,
|
|
)
|
|
)
|
|
rc = self._ctx.cfg.research_collector
|
|
if rc.enabled:
|
|
jobs.append(
|
|
JobSpec(
|
|
name="option_chain_research",
|
|
cron=rc.cron,
|
|
coro_factory=_option_chain_research,
|
|
)
|
|
)
|
|
_log.info(
|
|
"research collector ENABLED (cron=%s, window<=%dd, depth=%s)",
|
|
rc.cron,
|
|
rc.expiry_max_days,
|
|
rc.fetch_book_depth,
|
|
)
|
|
else:
|
|
_log.warning(
|
|
"data analysis disabled (CERBERO_BITE_ENABLE_DATA_ANALYSIS="
|
|
"false): market_snapshot job is NOT scheduled"
|
|
)
|
|
|
|
self._scheduler = build_scheduler(jobs)
|
|
return self._scheduler
|
|
|
|
async def run_forever(self, *, lock_path: Path | None = None) -> None:
|
|
"""Boot, acquire the single-instance lock, install the scheduler.
|
|
|
|
``lock_path`` defaults to ``<db_path.parent>/.lockfile`` so two
|
|
containers cannot trade against the same SQLite file. SIGTERM
|
|
and SIGINT are intercepted so Docker (or the operator) can
|
|
signal a clean shutdown — the scheduler is stopped, in-flight
|
|
cycles complete, the audit log fsyncs, and the HTTP client is
|
|
closed before the process exits.
|
|
"""
|
|
import signal # noqa: PLC0415 — only needed by run_forever
|
|
|
|
lock = EngineLock(
|
|
lock_path or self._ctx.db_path.parent / ".lockfile"
|
|
)
|
|
with lock:
|
|
try:
|
|
await self.boot()
|
|
scheduler = self.install_scheduler()
|
|
scheduler.start()
|
|
stop_event = asyncio.Event()
|
|
|
|
def _on_signal(signame: str) -> None:
|
|
_log.info("received %s — initiating shutdown", signame)
|
|
stop_event.set()
|
|
|
|
loop = asyncio.get_running_loop()
|
|
for sig_name in ("SIGTERM", "SIGINT"):
|
|
sig = getattr(signal, sig_name, None)
|
|
if sig is None: # pragma: no cover — Windows fallback
|
|
continue
|
|
try:
|
|
loop.add_signal_handler(
|
|
sig, _on_signal, sig_name
|
|
)
|
|
except NotImplementedError: # pragma: no cover
|
|
# Some sandboxes (Windows asyncio) don't support
|
|
# add_signal_handler; fall back to no-op.
|
|
signal.signal(sig, lambda *_: stop_event.set())
|
|
|
|
try:
|
|
await stop_event.wait()
|
|
finally:
|
|
scheduler.shutdown(wait=False)
|
|
finally:
|
|
await self._ctx.aclose()
|
|
|
|
|
|
def _run_backup(
|
|
*, db_path: Path, backup_dir: Path, retention_days: int
|
|
) -> None:
|
|
"""Synchronous helper invoked from the scheduler via ``asyncio.to_thread``.
|
|
|
|
Keeps the import of ``scripts.backup`` lazy: importing the module
|
|
eagerly at orchestrator load time would mean the scheduler depends
|
|
on a script that lives outside the ``cerbero_bite`` package, which
|
|
breaks ``importlib.util.spec_from_file_location`` if the cwd shifts
|
|
at runtime.
|
|
"""
|
|
import sys # noqa: PLC0415 — kept lazy to keep module load cheap
|
|
from importlib.util import ( # noqa: PLC0415
|
|
module_from_spec,
|
|
spec_from_file_location,
|
|
)
|
|
|
|
backup_py = Path(__file__).resolve().parents[3] / "scripts" / "backup.py"
|
|
spec = spec_from_file_location("_cerbero_bite_backup", backup_py)
|
|
if spec is None or spec.loader is None: # pragma: no cover — only on broken installs
|
|
raise RuntimeError(f"cannot load scripts/backup.py from {backup_py}")
|
|
module = module_from_spec(spec)
|
|
sys.modules.setdefault(spec.name, module)
|
|
spec.loader.exec_module(module)
|
|
module.backup_database(db_path=db_path, backup_dir=backup_dir)
|
|
module.prune_backups(backup_dir, retention_days=retention_days)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Convenience builder for the CLI
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def make_orchestrator(
|
|
*,
|
|
cfg: StrategyConfig,
|
|
endpoints: McpEndpoints,
|
|
token: str,
|
|
db_path: Path,
|
|
audit_path: Path,
|
|
expected_environment: Environment,
|
|
eur_to_usd: Decimal,
|
|
bot_tag: str | None = None,
|
|
flags: RuntimeFlags | None = None,
|
|
clock: Callable[[], datetime] | None = None,
|
|
) -> Orchestrator:
|
|
"""Build a fresh :class:`Orchestrator` ready for ``boot``/``run_*``."""
|
|
build_kwargs: dict[str, object] = {
|
|
"cfg": cfg,
|
|
"endpoints": endpoints,
|
|
"token": token,
|
|
"db_path": db_path,
|
|
"audit_path": audit_path,
|
|
"clock": clock or (lambda: datetime.now(UTC)),
|
|
}
|
|
if bot_tag is not None:
|
|
build_kwargs["bot_tag"] = bot_tag
|
|
ctx = build_runtime(**build_kwargs) # type: ignore[arg-type]
|
|
return Orchestrator(
|
|
ctx,
|
|
expected_environment=expected_environment,
|
|
eur_to_usd=eur_to_usd,
|
|
flags=flags,
|
|
)
|