diff --git a/Dockerfile b/Dockerfile index 6c7ca4b..0e6541a 100644 --- a/Dockerfile +++ b/Dockerfile @@ -1,4 +1,4 @@ -FROM python:3.12-slim +FROM python:3.12-slim-bookworm ENV PYTHONUNBUFFERED=1 \ PIP_NO_CACHE_DIR=1 \ @@ -12,8 +12,15 @@ RUN apt-get update \ COPY requirements.txt ./ RUN python -m pip install --upgrade pip \ - && python -m pip install -r requirements.txt + && python -m pip install -r requirements.txt \ + && apt-get purge -y git \ + && apt-get autoremove -y --purge \ + && rm -rf /var/lib/apt/lists/* COPY . . +RUN useradd --create-home --uid 1000 appuser \ + && chown -R appuser:appuser /app +USER appuser + CMD ["gunicorn", "--bind", ":8080", "main:app"] diff --git a/application/firstrade_client.py b/application/firstrade_client.py index d5847c8..2df1077 100644 --- a/application/firstrade_client.py +++ b/application/firstrade_client.py @@ -52,15 +52,23 @@ class FirstradeCredentials: @classmethod def from_env(cls, env: Callable[[str, str | None], str | None] = os.getenv) -> "FirstradeCredentials": - username = env("FIRSTRADE_USERNAME", "") or "" - password = env("FIRSTRADE_PASSWORD", "") or "" + def _get_credential(secret_name: str, env_var: str) -> str: + try: + from quant_platform_kit.cloud import get_secret_store + + return get_secret_store().get_secret(secret_name, project_id="firstradequant") + except Exception: + return env(env_var, "") or "" + + username = _get_credential("firstrade-username", "FIRSTRADE_USERNAME") + password = _get_credential("firstrade-password", "FIRSTRADE_PASSWORD") return cls( username=username.strip(), password=password, pin=env("FIRSTRADE_PIN", "") or "", email=env("FIRSTRADE_MFA_EMAIL", "") or "", phone=env("FIRSTRADE_MFA_PHONE", "") or "", - mfa_secret=env("FIRSTRADE_MFA_SECRET", "") or "", + mfa_secret=_get_credential("firstrade-mfa-secret", "FIRSTRADE_MFA_SECRET"), mfa_code=env("FIRSTRADE_MFA_CODE", "") or "", cookie_dir=env("FIRSTRADE_COOKIE_DIR", ".runtime/firstrade-cookies") or ".runtime/firstrade-cookies", diff --git a/application/monitor_dispatcher.py b/application/monitor_dispatcher.py deleted file mode 100644 index 1d59160..0000000 --- a/application/monitor_dispatcher.py +++ /dev/null @@ -1,271 +0,0 @@ -"""Dispatch shared monitor checks to platform Cloud Run services.""" - -from __future__ import annotations - -import datetime as dt -import json -import os -from concurrent.futures import ThreadPoolExecutor, as_completed -from dataclasses import dataclass -from typing import Any, Callable, Iterable, Mapping, Sequence -from zoneinfo import ZoneInfo - -import requests -from google.auth.transport.requests import Request -from google.oauth2 import id_token - - -MONITOR_TARGET_ENV_NAMES = ( - "MONITOR_DISPATCH_TARGETS_JSON", - "LONGBRIDGE_MONITOR_DISPATCH_TARGETS_JSON", - "SCHWAB_MONITOR_DISPATCH_TARGETS_JSON", - "FIRSTRADE_MONITOR_DISPATCH_TARGETS_JSON", -) -DEFAULT_LOOKBACK_MINUTES = 4 -DEFAULT_TIMEOUT_SECONDS = 120 -DEFAULT_MAX_WORKERS = 4 - - -@dataclass(frozen=True) -class MonitorWindow: - name: str - path: str - scheduler_key: str - - -MONITOR_WINDOWS = ( - MonitorWindow("probe", "/probe", "probe_time"), - MonitorWindow("precheck", "/dry-run", "precheck_time"), -) - - -def load_monitor_targets(env: Mapping[str, str] | None = None) -> list[dict[str, Any]]: - env = env or os.environ - raw = "" - for name in MONITOR_TARGET_ENV_NAMES: - raw = str(env.get(name) or "").strip() - if raw: - break - if not raw: - return [] - payload = json.loads(raw) - if isinstance(payload, dict): - targets = payload.get("targets") - else: - targets = payload - if not isinstance(targets, list): - raise ValueError("monitor dispatch targets must be a JSON array or an object with targets") - return [dict(target) for target in targets if isinstance(target, Mapping)] - - -def dispatch_due_monitors( - targets: Sequence[Mapping[str, Any]], - *, - now: dt.datetime | None = None, - lookback_minutes: int = DEFAULT_LOOKBACK_MINUTES, - timeout_seconds: int = DEFAULT_TIMEOUT_SECONDS, - max_workers: int = DEFAULT_MAX_WORKERS, - token_fetcher: Callable[[str], str] | None = None, - post_fn: Callable[..., Any] | None = None, -) -> dict[str, Any]: - now_utc = _as_utc(now or dt.datetime.now(dt.timezone.utc)) - due_dispatches = list(_iter_due_dispatches(targets, now_utc=now_utc, lookback_minutes=lookback_minutes)) - token_fetcher = token_fetcher or _fetch_id_token - post_fn = post_fn or requests.post - if not due_dispatches: - return { - "ok": True, - "total_targets": len(targets), - "dispatches_due": 0, - "results": [], - } - - workers = max(1, min(int(max_workers or 1), len(due_dispatches))) - results: list[dict[str, Any]] = [] - with ThreadPoolExecutor(max_workers=workers) as executor: - futures = [ - executor.submit( - _send_dispatch, - dispatch, - timeout_seconds=timeout_seconds, - token_fetcher=token_fetcher, - post_fn=post_fn, - ) - for dispatch in due_dispatches - ] - for future in as_completed(futures): - results.append(future.result()) - results.sort(key=lambda item: (str(item.get("service_name") or ""), str(item.get("window") or ""))) - return { - "ok": all(bool(result.get("ok")) for result in results), - "total_targets": len(targets), - "dispatches_due": len(due_dispatches), - "results": results, - } - - -def _iter_due_dispatches( - targets: Sequence[Mapping[str, Any]], - *, - now_utc: dt.datetime, - lookback_minutes: int, -) -> Iterable[dict[str, Any]]: - for target in targets: - if not _target_enabled(target): - continue - service_url = str(target.get("service_url") or "").rstrip("/") - if not service_url: - continue - scheduler = target.get("scheduler") if isinstance(target.get("scheduler"), Mapping) else {} - timezone = _target_timezone(scheduler) - local_now = now_utc.astimezone(timezone) - for window in MONITOR_WINDOWS: - schedule = str(scheduler.get(window.scheduler_key) or "").strip() - if not schedule: - continue - if _cron_due_within_window(schedule, local_now=local_now, lookback_minutes=lookback_minutes): - yield { - "service_name": str(target.get("service_name") or ""), - "strategy_profile": str(target.get("strategy_profile") or ""), - "account_scope": str(target.get("account_scope") or target.get("account_group") or ""), - "window": window.name, - "url": f"{service_url}{window.path}", - "audience": service_url, - } - - -def _send_dispatch( - dispatch: Mapping[str, Any], - *, - timeout_seconds: int, - token_fetcher: Callable[[str], str], - post_fn: Callable[..., Any], -) -> dict[str, Any]: - url = str(dispatch.get("url") or "") - audience = str(dispatch.get("audience") or "") - base_result = { - "service_name": dispatch.get("service_name"), - "strategy_profile": dispatch.get("strategy_profile"), - "account_scope": dispatch.get("account_scope"), - "window": dispatch.get("window"), - "url": url, - } - try: - token = token_fetcher(audience) - response = post_fn( - url, - headers={ - "Authorization": f"Bearer {token}", - "User-Agent": "platform-monitor-dispatcher", - }, - timeout=timeout_seconds, - ) - status_code = int(getattr(response, "status_code", 0) or 0) - return { - **base_result, - "status_code": status_code, - "ok": 200 <= status_code < 300, - } - except Exception as exc: - return { - **base_result, - "status_code": 0, - "ok": False, - "error_type": type(exc).__name__, - "error": str(exc), - } - - -def _target_enabled(target: Mapping[str, Any]) -> bool: - value = target.get("runtime_target_enabled") - if value is None: - value = target.get("enabled") - if value is None: - return True - if isinstance(value, bool): - return value - return str(value).strip().lower() not in {"0", "false", "no", "off", "disabled"} - - -def _target_timezone(scheduler: Mapping[str, Any]) -> dt.tzinfo: - try: - return ZoneInfo(str(scheduler.get("timezone") or "UTC")) - except Exception: - return dt.timezone.utc - - -def _cron_due_within_window(schedule: str, *, local_now: dt.datetime, lookback_minutes: int) -> bool: - fields = schedule.split() - if len(fields) != 5: - return False - lookback = max(0, int(lookback_minutes or 0)) - floor_now = local_now.replace(second=0, microsecond=0) - for minute_offset in range(lookback + 1): - candidate = floor_now - dt.timedelta(minutes=minute_offset) - if _cron_matches(fields, candidate): - return True - return False - - -def _cron_matches(fields: Sequence[str], value: dt.datetime) -> bool: - minute, hour, day_of_month, month, day_of_week = fields - cron_weekday = (value.weekday() + 1) % 7 - return ( - _field_matches(minute, value.minute, 0, 59) - and _field_matches(hour, value.hour, 0, 23) - and _field_matches(day_of_month, value.day, 1, 31) - and _field_matches(month, value.month, 1, 12) - and _field_matches(day_of_week, cron_weekday, 0, 7, sunday_alias=True) - ) - - -def _field_matches(field: str, value: int, min_value: int, max_value: int, *, sunday_alias: bool = False) -> bool: - for part in field.split(","): - part = part.strip() - if not part: - continue - if _part_matches(part, value, min_value, max_value, sunday_alias=sunday_alias): - return True - return False - - -def _part_matches(part: str, value: int, min_value: int, max_value: int, *, sunday_alias: bool = False) -> bool: - if "/" in part: - base, step_text = part.split("/", 1) - try: - step = int(step_text) - except ValueError: - return False - if step <= 0: - return False - else: - base = part - step = 1 - if base == "*": - start, end = min_value, max_value - elif "-" in base: - start_text, end_text = base.split("-", 1) - try: - start, end = int(start_text), int(end_text) - except ValueError: - return False - else: - try: - start = end = int(base) - except ValueError: - return False - if sunday_alias and value == 0 and start == end == 7: - return True - if value < start or value > end: - return False - return (value - start) % step == 0 - - -def _as_utc(value: dt.datetime) -> dt.datetime: - if value.tzinfo is None: - return value.replace(tzinfo=dt.timezone.utc) - return value.astimezone(dt.timezone.utc) - - -def _fetch_id_token(audience: str) -> str: - return id_token.fetch_id_token(Request(), audience) diff --git a/application/state_persistence.py b/application/state_persistence.py index 36fa480..255fd09 100644 --- a/application/state_persistence.py +++ b/application/state_persistence.py @@ -1,9 +1,4 @@ -"""Lightweight runtime state persistence. - -The service intentionally avoids heavyweight Google client libraries. Cloud Run -already has a metadata server token, and the JSON APIs are enough for small -session and account-state payloads. -""" +"""Lightweight runtime state persistence using quant_platform_kit.cloud object store.""" from __future__ import annotations @@ -11,13 +6,6 @@ import os from dataclasses import dataclass from typing import Any, Callable -from urllib.parse import quote, urlencode - -import requests - -METADATA_TOKEN_URL = ( - "http://metadata.google.internal/computeMetadata/v1/instance/service-accounts/default/token" -) class StatePersistenceError(RuntimeError): @@ -28,84 +16,60 @@ def _normalize_prefix(prefix: str | None) -> str: return str(prefix or "").strip("/") +def _object_store(): + try: + from quant_platform_kit.cloud import get_object_store + + return get_object_store() + except ImportError: + raise StatePersistenceError("quant_platform_kit.cloud not available") + + @dataclass(frozen=True) class GcsStateStore: bucket: str prefix: str = "firstrade-platform" - timeout_seconds: float = 10.0 - token_getter: Callable[[], str] | None = None def __post_init__(self) -> None: if not str(self.bucket or "").strip(): raise ValueError("GCS state bucket must be non-empty.") + def _object_uri(self, key: str) -> str: + clean_key = str(key or "").strip("/") + if not clean_key: + raise ValueError("State key must be non-empty.") + prefix = _normalize_prefix(self.prefix) + object_name = f"{prefix}/{clean_key}" if prefix else clean_key + return f"gs://{self.bucket}/{object_name}" + def read_json(self, key: str) -> dict[str, Any] | None: - object_name = self._object_name(key) - url = ( - f"https://storage.googleapis.com/storage/v1/b/{quote(self.bucket, safe='')}/o/" - f"{quote(object_name, safe='')}?alt=media" - ) - response = requests.get( - url, - headers=self._auth_headers(), - timeout=self.timeout_seconds, - ) - if response.status_code == 404: + uri = self._object_uri(key) + try: + text = _object_store().read_text(uri) + except Exception as exc: + error_msg = str(exc) + if "404" in error_msg or "NOT_FOUND" in error_msg.upper(): + return None + raise StatePersistenceError(f"GCS read failed for {key}: {exc}") from exc + if text is None: return None - if response.status_code != 200: - raise StatePersistenceError( - f"GCS read failed for {object_name}: HTTP {response.status_code}" - ) - payload = response.json() + try: + payload = json.loads(text) + except json.JSONDecodeError as exc: + raise StatePersistenceError(f"GCS payload for {key} is not valid JSON: {exc}") from exc if not isinstance(payload, dict): - raise StatePersistenceError(f"GCS payload for {object_name} is not a JSON object.") + raise StatePersistenceError(f"GCS payload for {key} is not a JSON object.") return payload def write_json(self, key: str, payload: dict[str, Any]) -> bool: - object_name = self._object_name(key) - query = urlencode({"uploadType": "media", "name": object_name}) - url = f"https://storage.googleapis.com/upload/storage/v1/b/{quote(self.bucket, safe='')}/o?{query}" - response = requests.post( - url, - headers={ - **self._auth_headers(), - "Content-Type": "application/json; charset=utf-8", - }, - data=json.dumps(payload, ensure_ascii=False, separators=(",", ":")).encode("utf-8"), - timeout=self.timeout_seconds, - ) - if response.status_code not in (200, 201): - raise StatePersistenceError( - f"GCS write failed for {object_name}: HTTP {response.status_code}" - ) + uri = self._object_uri(key) + data = json.dumps(payload, ensure_ascii=False, separators=(",", ":")) + try: + _object_store().write_text(uri, data) + except Exception as exc: + raise StatePersistenceError(f"GCS write failed for {key}: {exc}") from exc return True - def _object_name(self, key: str) -> str: - clean_key = str(key or "").strip("/") - if not clean_key: - raise ValueError("State key must be non-empty.") - prefix = _normalize_prefix(self.prefix) - return f"{prefix}/{clean_key}" if prefix else clean_key - - def _auth_headers(self) -> dict[str, str]: - token = self.token_getter() if self.token_getter is not None else _metadata_access_token() - return {"Authorization": f"Bearer {token}"} - - -def _metadata_access_token() -> str: - response = requests.get( - METADATA_TOKEN_URL, - headers={"Metadata-Flavor": "Google"}, - timeout=3.0, - ) - if response.status_code != 200: - raise StatePersistenceError(f"Metadata token request failed: HTTP {response.status_code}") - payload = response.json() - token = payload.get("access_token") - if not token: - raise StatePersistenceError("Metadata token response did not include access_token.") - return str(token) - def build_gcs_state_store_from_env( env: Callable[[str, str | None], str | None] = os.getenv, diff --git a/main.py b/main.py index b869dab..27c5505 100644 --- a/main.py +++ b/main.py @@ -10,7 +10,7 @@ from flask import Flask, jsonify, request -from application.monitor_dispatcher import dispatch_due_monitors, load_monitor_targets +from quant_platform_kit.common.platform_runner import dispatch_due_monitors, load_monitor_targets from application.firstrade_client import ( FirstradeBrokerClient, FirstradeCredentials, @@ -57,9 +57,18 @@ def _split_env_list(value: str | None) -> tuple[str, ...]: ) +def _get_telegram_token() -> str: + try: + from quant_platform_kit.cloud import get_secret_store + + return get_secret_store().get_secret("firstrade-telegram-token", project_id="firstradequant") + except Exception: + return os.environ.get("TELEGRAM_TOKEN", "") + + def _telegram_notification_targets() -> tuple[tuple[str, str], ...]: targets: list[tuple[str, str]] = [] - main_token = os.getenv("TELEGRAM_TOKEN") + main_token = _get_telegram_token() main_chat_id = os.getenv("GLOBAL_TELEGRAM_CHAT_ID") if main_token and main_chat_id: targets.append((main_token, main_chat_id)) diff --git a/pyproject.toml b/pyproject.toml index 2f0d36f..33efb59 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -16,7 +16,7 @@ dependencies = [ "firstrade==0.0.39", "pandas_market_calendars", "pytz", - "quant-platform-kit @ git+https://github.com/QuantStrategyLab/QuantPlatformKit.git@dfdbef6b58ab46f357d67800510bb9e8c4a01182", + "quant-platform-kit @ git+https://github.com/QuantStrategyLab/QuantPlatformKit.git@e86554b", "us-equity-strategies @ git+https://github.com/QuantStrategyLab/UsEquityStrategies.git@a9546362a27abdfc9cd5184b30ca8d26fd774187", "google-cloud-storage", "requests", diff --git a/requirements.txt b/requirements.txt index 20537d7..d4629a5 100644 --- a/requirements.txt +++ b/requirements.txt @@ -3,7 +3,7 @@ gunicorn firstrade==0.0.39 pandas_market_calendars pytz -quant-platform-kit @ git+https://github.com/QuantStrategyLab/QuantPlatformKit.git@dfdbef6b58ab46f357d67800510bb9e8c4a01182 +quant-platform-kit @ git+https://github.com/QuantStrategyLab/QuantPlatformKit.git@e86554b us-equity-strategies @ git+https://github.com/QuantStrategyLab/UsEquityStrategies.git@a9546362a27abdfc9cd5184b30ca8d26fd774187 google-cloud-storage google-auth diff --git a/runtime_config_support.py b/runtime_config_support.py index 963deea..7aed58b 100644 --- a/runtime_config_support.py +++ b/runtime_config_support.py @@ -31,6 +31,18 @@ IBIT_SMART_DCA_PROFILE = "ibit_smart_dca" +def _get_credential(secret_name: str, env_var: str) -> str: + try: + from quant_platform_kit.cloud import get_secret_store + + return get_secret_store().get_secret(secret_name, project_id="firstradequant") + except Exception: + val = os.environ.get(env_var) + if val is None: + raise ValueError(f"Missing credential: {secret_name} (tried env {env_var})") + return val + + @dataclass(frozen=True) class PlatformRuntimeSettings: project_id: str | None @@ -162,7 +174,7 @@ def load_platform_runtime_settings( strategy_display_name=runtime_paths.strategy_display_name, strategy_domain=runtime_paths.strategy_domain, notify_lang=os.getenv("NOTIFY_LANG", "en"), - tg_token=os.getenv("TELEGRAM_TOKEN"), + tg_token=_get_credential("firstrade-telegram-token", "TELEGRAM_TOKEN"), tg_chat_id=os.getenv("GLOBAL_TELEGRAM_CHAT_ID"), dry_run_only=dry_run_only, runtime_target_enabled=_runtime_target_enabled_env(),