Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
145 changes: 145 additions & 0 deletions application/startup_config.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,145 @@
"""Explicit startup config normalization for LongBridge (B08, batch 1).

Pure functions: they never read or write ``os.environ``. ``main`` applies the
results explicitly (settings replacement + an env overlay for the monitor
dispatcher) and keeps the returned corrections as an adoption receipt.

Normalization semantics are unchanged from the previous inline module-level
code in ``main.py``:
- plugin mount ``strategy`` fields are rewritten to the active STRATEGY_PROFILE;
- monitor dispatch target ``strategy_profile`` fields are rewritten to the
active STRATEGY_PROFILE;
- malformed / non-matching JSON is left untouched; unchanged input is returned
verbatim (no re-serialization).
"""

from __future__ import annotations

import json
from collections.abc import Mapping
from dataclasses import dataclass
from typing import Callable

MONITOR_DISPATCH_TARGET_ENV_NAMES = (
"MONITOR_DISPATCH_TARGETS_JSON",
"LONGBRIDGE_MONITOR_DISPATCH_TARGETS_JSON",
)


@dataclass(frozen=True)
class ConfigCorrection:
"""One adopted correction: which source/field changed and from/to what."""

source: str
field: str
original: str
adopted: str


@dataclass(frozen=True)
class StartupConfigReceipt:
"""Receipt of the values adopted at startup (no secrets; profile names only)."""

strategy_profile: str
corrections: tuple[ConfigCorrection, ...] = ()
env_overrides: tuple[str, ...] = ()

def as_dict(self) -> dict[str, object]:
return {
"strategy_profile": self.strategy_profile,
"corrections": [
{"source": c.source, "field": c.field, "original": c.original, "adopted": c.adopted}
for c in self.corrections
],
"env_overrides": list(self.env_overrides),
}


def _default_log(message: str) -> None:
print(message, flush=True)


def normalize_plugin_mounts_strategy(
raw_mounts_json: str | None,
*,
strategy_profile: str,
log: Callable[[str], None] = _default_log,
) -> tuple[str | None, tuple[ConfigCorrection, ...]]:
if not raw_mounts_json:
return raw_mounts_json, ()
try:
mounts = json.loads(raw_mounts_json)
except json.JSONDecodeError:
return raw_mounts_json, ()
if not isinstance(mounts, dict):
return raw_mounts_json, ()
plugins = mounts.get("strategy_plugins")
if not isinstance(plugins, list):
return raw_mounts_json, ()
corrections: list[ConfigCorrection] = []
for entry in plugins:
if not isinstance(entry, dict):
continue
old = entry.get("strategy")
if old and old != strategy_profile:
entry["strategy"] = strategy_profile
corrections.append(
ConfigCorrection(
source="STRATEGY_PLUGIN_MOUNTS_JSON",
field="strategy_plugins[].strategy",
original=str(old),
adopted=strategy_profile,
)
)
log(f"[config-sync] Plugin mount strategy corrected: {old} → {strategy_profile}")
if not corrections:
return raw_mounts_json, ()
return json.dumps(mounts, ensure_ascii=False), tuple(corrections)


def normalize_monitor_dispatch_targets(
env: Mapping[str, str],
*,
strategy_profile: str,
log: Callable[[str], None] = _default_log,
) -> tuple[dict[str, str], tuple[ConfigCorrection, ...]]:
"""Return ``{env_name: corrected_json}`` for changed names only (env untouched)."""
overrides: dict[str, str] = {}
corrections: list[ConfigCorrection] = []
for env_name in MONITOR_DISPATCH_TARGET_ENV_NAMES:
raw = env.get(env_name)
if not raw:
continue
try:
targets = json.loads(raw)
except json.JSONDecodeError:
continue
if not isinstance(targets, dict):
continue
entries = targets.get("targets")
if not isinstance(entries, list):
continue
changed = False
for entry in entries:
if isinstance(entry, dict):
old = entry.get("strategy_profile")
if old and old != strategy_profile:
entry["strategy_profile"] = strategy_profile
changed = True
corrections.append(
ConfigCorrection(
source=env_name,
field="targets[].strategy_profile",
original=str(old),
adopted=strategy_profile,
)
)
log(f"[config-sync] {env_name} strategy_profile corrected: {old} → {strategy_profile}")
if changed:
overrides[env_name] = json.dumps(targets, ensure_ascii=False)
return overrides, tuple(corrections)


def overlay_env(env: Mapping[str, str], overrides: Mapping[str, str]) -> dict[str, str]:
"""Read-only view used by consumers that previously read the mutated env."""
return {**dict(env), **dict(overrides)}
109 changes: 58 additions & 51 deletions main.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,7 @@
Runs the configured shared UsEquityStrategies profile for the deployed LongBridge account.
Runs on Cloud Run; token from Secret Manager, orders via LongPort OpenAPI, alerts via Telegram.
"""
import dataclasses
import json
import importlib
import os
Expand All @@ -14,6 +15,12 @@
import google.auth
import requests
from application.monitor_dispatcher import dispatch_due_monitors, load_monitor_targets
from application.startup_config import (
StartupConfigReceipt,
normalize_monitor_dispatch_targets,
normalize_plugin_mounts_strategy,
overlay_env,
)
from application.broker_reconciliation import (
account_snapshot_enabled,
collect_read_only_reconciliation_observations,
Expand Down Expand Up @@ -102,56 +109,56 @@ def get_project_id():


def _normalize_plugin_mounts_strategy(raw_mounts_json: str | None) -> str | None:
if not raw_mounts_json:
return raw_mounts_json
try:
mounts = json.loads(raw_mounts_json)
except json.JSONDecodeError:
return raw_mounts_json
if not isinstance(mounts, dict):
return raw_mounts_json
plugins = mounts.get("strategy_plugins")
if not isinstance(plugins, list):
return raw_mounts_json
changed = False
for entry in plugins:
if not isinstance(entry, dict):
continue
old = entry.get("strategy")
if old and old != STRATEGY_PROFILE:
entry["strategy"] = STRATEGY_PROFILE
changed = True
print(f"[config-sync] Plugin mount strategy corrected: {old} → {STRATEGY_PROFILE}", flush=True)
return json.dumps(mounts, ensure_ascii=False) if changed else raw_mounts_json


if hasattr(RUNTIME_SETTINGS, "strategy_plugin_mounts_json"):
normalized = _normalize_plugin_mounts_strategy(RUNTIME_SETTINGS.strategy_plugin_mounts_json)
if normalized != RUNTIME_SETTINGS.strategy_plugin_mounts_json:
object.__setattr__(RUNTIME_SETTINGS, "strategy_plugin_mounts_json", normalized)

for _env_name in ("MONITOR_DISPATCH_TARGETS_JSON", "LONGBRIDGE_MONITOR_DISPATCH_TARGETS_JSON"):
_raw = os.environ.get(_env_name)
if _raw:
try:
_targets = json.loads(_raw)
except json.JSONDecodeError:
continue
if not isinstance(_targets, dict):
continue
_entries = _targets.get("targets")
if not isinstance(_entries, list):
continue
_changed = False
for _entry in _entries:
if isinstance(_entry, dict):
_old = _entry.get("strategy_profile")
if _old and _old != STRATEGY_PROFILE:
_entry["strategy_profile"] = STRATEGY_PROFILE
_changed = True
print(f"[config-sync] {_env_name} strategy_profile corrected: {_old} → {STRATEGY_PROFILE}", flush=True)
if _changed:
os.environ[_env_name] = json.dumps(_targets, ensure_ascii=False)
normalized, _corrections = normalize_plugin_mounts_strategy(
raw_mounts_json,
strategy_profile=STRATEGY_PROFILE,
)
return normalized


def _adopt_startup_config(settings, env):
"""Explicitly normalize startup config without mutating ``os.environ`` (B08).

Returns ``(settings, monitor_dispatch_env_overrides, receipt)``.
"""
corrections = []
if hasattr(settings, "strategy_plugin_mounts_json"):
normalized, plugin_corrections = normalize_plugin_mounts_strategy(
settings.strategy_plugin_mounts_json,
strategy_profile=STRATEGY_PROFILE,
)
corrections.extend(plugin_corrections)
if normalized != settings.strategy_plugin_mounts_json:
if dataclasses.is_dataclass(settings) and not isinstance(settings, type):
settings = dataclasses.replace(settings, strategy_plugin_mounts_json=normalized)
else: # non-dataclass settings objects (test doubles)
setattr(settings, "strategy_plugin_mounts_json", normalized)
overrides, monitor_corrections = normalize_monitor_dispatch_targets(
env,
strategy_profile=STRATEGY_PROFILE,
)
corrections.extend(monitor_corrections)
receipt = StartupConfigReceipt(
strategy_profile=STRATEGY_PROFILE,
corrections=tuple(corrections),
env_overrides=tuple(sorted(overrides)),
)
return settings, overrides, receipt


RUNTIME_SETTINGS, MONITOR_DISPATCH_ENV_OVERRIDES, STARTUP_CONFIG_RECEIPT = _adopt_startup_config(
RUNTIME_SETTINGS,
os.environ,
)


def _load_monitor_dispatch_targets():
# No corrections: identical call to before. Corrections: read through an
# explicit overlay instead of the previously mutated ``os.environ``.
if not MONITOR_DISPATCH_ENV_OVERRIDES:
return load_monitor_targets()
return load_monitor_targets(overlay_env(os.environ, MONITOR_DISPATCH_ENV_OVERRIDES))


ACCOUNT_REGION = RUNTIME_SETTINGS.account_region
MARKET = RUNTIME_SETTINGS.market
Expand Down Expand Up @@ -1415,7 +1422,7 @@ def handle_monitor_dispatch():
if request_method() == "GET":
return "Monitor Dispatch OK - use POST to dispatch due monitor checks", 200
try:
result = dispatch_due_monitors(load_monitor_targets())
result = dispatch_due_monitors(_load_monitor_dispatch_targets())
except Exception as exc:
return _handle_route_runtime_error(exc, route_label="POST /monitor-dispatch")
return json.dumps(result, ensure_ascii=False), 200, {"Content-Type": "application/json"}
Expand Down
Loading
Loading