Add Firstrade session reuse and align notifications (#11)

Author: Pigbibi

.github/workflows/sync-cloud-run-env.yml

@@ -39,6 +39,8 @@ jobs:
       FIRSTRADE_ACCOUNT: ${{ vars.FIRSTRADE_ACCOUNT }}
       FIRSTRADE_COOKIE_DIR: ${{ vars.FIRSTRADE_COOKIE_DIR }}
       FIRSTRADE_DRY_RUN_ONLY: ${{ vars.FIRSTRADE_DRY_RUN_ONLY }}
+      FIRSTRADE_REUSE_SESSION: ${{ vars.FIRSTRADE_REUSE_SESSION }}
+      FIRSTRADE_SESSION_CACHE_TTL_SECONDS: ${{ vars.FIRSTRADE_SESSION_CACHE_TTL_SECONDS }}
       FIRSTRADE_ENABLE_LIVE_TRADING: ${{ vars.FIRSTRADE_ENABLE_LIVE_TRADING }}
       FIRSTRADE_RUN_SMOKE_ON_HTTP: ${{ vars.FIRSTRADE_RUN_SMOKE_ON_HTTP }}
       FIRSTRADE_RUN_STRATEGY_ON_HTTP: ${{ vars.FIRSTRADE_RUN_STRATEGY_ON_HTTP }}
@@ -390,6 +392,8 @@ jobs:
           add_optional_env FIRSTRADE_ACCOUNT
           add_optional_env FIRSTRADE_COOKIE_DIR
           add_optional_env FIRSTRADE_DRY_RUN_ONLY
+          add_optional_env FIRSTRADE_REUSE_SESSION
+          add_optional_env FIRSTRADE_SESSION_CACHE_TTL_SECONDS
           add_optional_env FIRSTRADE_ENABLE_LIVE_TRADING
           add_optional_env FIRSTRADE_RUN_SMOKE_ON_HTTP
           add_optional_env FIRSTRADE_RUN_STRATEGY_ON_HTTP

README.md

@@ -71,6 +71,8 @@ commit credentials.
 | `FIRSTRADE_ACCOUNT` | Optional | Required when multiple accounts are returned |
 | `STRATEGY_PROFILE` | Yes for runtime | Shared US equity strategy profile |
 | `FIRSTRADE_DRY_RUN_ONLY` | Optional | Defaults to `true` for platform runtime |
+| `FIRSTRADE_REUSE_SESSION` | Optional | Reuse cached Firstrade session headers inside the same warm runtime instance before logging in again. Defaults to `false` |
+| `FIRSTRADE_SESSION_CACHE_TTL_SECONDS` | Optional | Max age for local session header reuse when `FIRSTRADE_REUSE_SESSION=true`. Defaults to `21600` |
 | `ACCOUNT_PREFIX` | Optional | Alert/log prefix, default `FIRSTRADE` |
 | `ACCOUNT_REGION` | Optional | Runtime account scope, default `US` |
 | `NOTIFY_LANG` | Optional | Notification language, `en` or `zh` |
@@ -173,6 +175,12 @@ The strategy execution service uses whole-share limit orders for generated
 strategy orders. If the notional cap is below the current price of a target
 symbol, that order is skipped instead of being enlarged.
 
+`FIRSTRADE_REUSE_SESSION=true` reduces repeated login attempts while the same
+Cloud Run instance stays warm. It stores the current session headers only in the
+container-local cookie directory and tries that session before calling Firstrade
+login again. A cold start, new revision, expired session, or broker-side
+invalidation still falls back to a fresh login.
+
 ## Cloud Run Shape
 
 `main.py` exposes:

application/execution_service.py

@@ -185,6 +185,8 @@ def _submit_order(
         "symbol": report.symbol,
         "side": report.side,
         "quantity": report.quantity,
+        "order_type": "limit",
+        "limit_price": round(float(limit_price), 2),
         "status": report.status,
         "broker_order_id": report.broker_order_id,
         "raw_payload": report.raw_payload,

application/firstrade_client.py

@@ -6,9 +6,11 @@
 
 from __future__ import annotations
 
+import json
 import os
 from dataclasses import dataclass
 from pathlib import Path
+from time import time
 from typing import Any, Callable
 
 
@@ -38,6 +40,8 @@ class FirstradeCredentials:
     mfa_secret: str = ""
     mfa_code: str = ""
     cookie_dir: str = ".runtime/firstrade-cookies"
+    reuse_session: bool = False
+    session_cache_ttl_seconds: int = 21_600
     debug: bool = False
 
     @classmethod
@@ -54,6 +58,11 @@ def from_env(cls, env: Callable[[str, str | None], str | None] = os.getenv) -> "
             mfa_code=env("FIRSTRADE_MFA_CODE", "") or "",
             cookie_dir=env("FIRSTRADE_COOKIE_DIR", ".runtime/firstrade-cookies")
             or ".runtime/firstrade-cookies",
+            reuse_session=(env("FIRSTRADE_REUSE_SESSION", "false") or "").strip().lower() == "true",
+            session_cache_ttl_seconds=_coerce_positive_int(
+                env("FIRSTRADE_SESSION_CACHE_TTL_SECONDS", "21600"),
+                default=21_600,
+            ),
             debug=(env("FIRSTRADE_DEBUG", "false") or "").lower() == "true",
         )
 
@@ -104,6 +113,14 @@ def _coerce_positive_float(value: float | None, field: str) -> float | None:
     return coerced
 
 
+def _coerce_positive_int(value: str | None, *, default: int) -> int:
+    try:
+        coerced = int(str(value or "").strip())
+    except ValueError:
+        return default
+    return coerced if coerced > 0 else default
+
+
 def validate_stock_order(
     request: StockOrderRequest,
     *,
@@ -188,6 +205,7 @@ def __init__(
         self._ohlc_factory = ohlc_factory
         self.session: Any | None = None
         self.account_data: Any | None = None
+        self.session_reused = False
 
     def connect(self) -> "FirstradeBrokerClient":
         self.credentials.require_login_fields()
@@ -201,16 +219,14 @@ def connect(self) -> "FirstradeBrokerClient":
 
         cookie_dir = Path(self.credentials.cookie_dir)
         cookie_dir.mkdir(parents=True, exist_ok=True)
-        session = session_factory(
-            username=self.credentials.username,
-            password=self.credentials.password,
-            pin=self.credentials.pin,
-            email=self.credentials.email,
-            phone=self.credentials.phone,
-            mfa_secret=self.credentials.mfa_secret,
-            profile_path=str(cookie_dir),
-            debug=self.credentials.debug,
-        )
+        session = self._build_session(session_factory, cookie_dir)
+        if self.credentials.reuse_session and self._try_cached_session(
+            session,
+            account_data_factory=account_data_factory,
+            cookie_dir=cookie_dir,
+        ):
+            return self
+
         needs_mfa_code = bool(session.login())
         if needs_mfa_code:
             if not self.credentials.mfa_code:
@@ -220,8 +236,90 @@ def connect(self) -> "FirstradeBrokerClient":
             session.login_two(self.credentials.mfa_code)
         self.session = session
         self.account_data = account_data_factory(session)
+        self.session_reused = False
+        self._save_session_cache(cookie_dir)
         return self
 
+    def _build_session(self, session_factory: Callable[..., Any], cookie_dir: Path) -> Any:
+        return session_factory(
+            username=self.credentials.username,
+            password=self.credentials.password,
+            pin=self.credentials.pin,
+            email=self.credentials.email,
+            phone=self.credentials.phone,
+            mfa_secret=self.credentials.mfa_secret,
+            profile_path=str(cookie_dir),
+            debug=self.credentials.debug,
+        )
+
+    def _session_cache_path(self, cookie_dir: Path) -> Path:
+        safe_username = "".join(ch for ch in self.credentials.username if ch.isalnum() or ch in ("-", "_"))
+        return cookie_dir / f"ft_session{safe_username}.json"
+
+    def _load_session_cache(self, cookie_dir: Path) -> dict[str, Any] | None:
+        path = self._session_cache_path(cookie_dir)
+        try:
+            payload = json.loads(path.read_text())
+        except (OSError, json.JSONDecodeError):
+            return None
+        if not isinstance(payload, dict):
+            return None
+        try:
+            saved_at = float(payload.get("saved_at") or 0.0)
+        except (TypeError, ValueError):
+            return None
+        ttl = max(1, int(self.credentials.session_cache_ttl_seconds or 1))
+        if saved_at <= 0.0 or (time() - saved_at) > ttl:
+            return None
+        if not payload.get("ftat") or not payload.get("sid"):
+            return None
+        return payload
+
+    def _try_cached_session(
+        self,
+        session: Any,
+        *,
+        account_data_factory: Callable[[Any], Any],
+        cookie_dir: Path,
+    ) -> bool:
+        payload = self._load_session_cache(cookie_dir)
+        if not payload:
+            return False
+        try:
+            from firstrade import urls
+
+            session.session.headers.update(urls.session_headers())
+            session.session.headers["access-token"] = urls.access_token()
+            session.session.headers["ftat"] = str(payload["ftat"])
+            session.session.headers["sid"] = str(payload["sid"])
+            account_data = account_data_factory(session)
+        except Exception:
+            try:
+                self._session_cache_path(cookie_dir).unlink()
+            except OSError:
+                pass
+            return False
+        self.session = session
+        self.account_data = account_data
+        self.session_reused = True
+        return True
+
+    def _save_session_cache(self, cookie_dir: Path) -> None:
+        if not self.credentials.reuse_session or self.session is None:
+            return
+        headers = getattr(getattr(self.session, "session", None), "headers", {}) or {}
+        payload = {
+            "ftat": headers.get("ftat"),
+            "sid": headers.get("sid"),
+            "saved_at": time(),
+        }
+        if not payload["ftat"] or not payload["sid"]:
+            return
+        try:
+            self._session_cache_path(cookie_dir).write_text(json.dumps(payload), encoding="utf-8")
+        except OSError:
+            return
+
     def require_connected(self) -> tuple[Any, Any]:
         if self.session is None or self.account_data is None:
             raise FirstradePlatformError("Firstrade client is not connected.")

application/rebalance_service.py

@@ -147,6 +147,7 @@ def run_strategy_cycle(
         live_trading_enabled=settings.live_trading_enabled,
         client_factory=client_factory,
     )
+    print(f"Firstrade session reused={bool(getattr(client, 'session_reused', False))}", flush=True)
     account = client.select_account(env_reader("FIRSTRADE_ACCOUNT", "") or None)
     strategy_runtime = load_strategy_runtime(
         settings.strategy_profile,
@@ -210,6 +211,7 @@ def run_strategy_cycle(
         "strategy_display_name": strategy_runtime.display_name,
         "dry_run_only": settings.dry_run_only,
         "live_trading_enabled": settings.live_trading_enabled,
+        "session_reused": bool(getattr(client, "session_reused", False)),
         "portfolio": plan.get("portfolio", {}),
         "allocation": plan.get("allocation", {}),
         "execution": plan.get("execution", {}),

notifications/telegram.py

@@ -33,6 +33,7 @@
         "investable_cash": "可投资现金",
         "holdings_title": "💼 策略持仓",
         "holding_line": "{symbol}: {market_value} / {quantity}",
+        "quantity_share": "{quantity}股",
         "quantity_shares": "{quantity}股",
         "signal_label": "信号",
         "separator": SEPARATOR,
@@ -43,10 +44,15 @@
         "market_status_line": "📊 市场状态: {status}",
         "signal_line": "🎯 信号: {signal}",
         "target_diff_summary": "调仓变化: {details}",
-        "dry_run_buy_order": "🧪 模拟买单: {symbol} {quantity}",
-        "dry_run_sell_order": "🧪 模拟卖单: {symbol} {quantity}",
-        "submitted_buy_order": "已提交买单: {symbol} {quantity}",
-        "submitted_sell_order": "已提交卖单: {symbol} {quantity}",
+        "order_logs_title": "🧾 执行明细",
+        "dry_run_order": "🧪 模拟{order_type}{side} {symbol}: {quantity}{price}",
+        "submitted_order": "{icon} 已提交{order_type}{side} {symbol}: {quantity}{price}{order_id}",
+        "order_type_limit": "限价",
+        "order_type_market": "市价",
+        "side_buy": "买入",
+        "side_sell": "卖出",
+        "order_price_suffix": " @ ${price}",
+        "order_id_suffix": "（订单号: {order_id}）",
         "no_order_submitted": "未下单: 原因={reason}",
         "no_rebalance_needed": "✅ 无需调仓",
         "no_trades": "✅ 无需调仓",
@@ -105,6 +111,7 @@
         "investable_cash": "Investable cash",
         "holdings_title": "💼 Strategy Holdings",
         "holding_line": "{symbol}: {market_value} / {quantity}",
+        "quantity_share": "{quantity} share",
         "quantity_shares": "{quantity} shares",
         "signal_label": "Signal",
         "separator": SEPARATOR,
@@ -115,10 +122,15 @@
         "market_status_line": "📊 Market: {status}",
         "signal_line": "🎯 Signal: {signal}",
         "target_diff_summary": "Target changes: {details}",
-        "dry_run_buy_order": "🧪 Dry-run buy: {symbol} {quantity}",
-        "dry_run_sell_order": "🧪 Dry-run sell: {symbol} {quantity}",
-        "submitted_buy_order": "Submitted buy: {symbol} {quantity}",
-        "submitted_sell_order": "Submitted sell: {symbol} {quantity}",
+        "order_logs_title": "🧾 Execution details",
+        "dry_run_order": "🧪 Dry-run {order_type} {side} {symbol}: {quantity}{price}",
+        "submitted_order": "{icon} Submitted {order_type} {side} {symbol}: {quantity}{price}{order_id}",
+        "order_type_limit": "limit",
+        "order_type_market": "market",
+        "side_buy": "buy",
+        "side_sell": "sell",
+        "order_price_suffix": " @ ${price}",
+        "order_id_suffix": " (ID: {order_id})",
         "no_order_submitted": "No order submitted: reason={reason}",
         "no_rebalance_needed": "✅ No rebalance needed",
         "no_trades": "✅ No rebalance needed",
@@ -211,6 +223,11 @@ def _format_money(value: Any) -> str:
     return "$0.00" if number is None else f"${number:,.2f}"
 
 
+def _format_price(value: Any) -> str:
+    number = _safe_float(value)
+    return "" if number is None else f"{number:,.2f}"
+
+
 def _format_quantity(value: Any) -> str:
     number = _safe_float(value)
     if number is None:
@@ -221,7 +238,9 @@ def _format_quantity(value: Any) -> str:
 
 
 def _format_shares(value: Any, *, translator: Callable[..., str]) -> str:
-    return translator("quantity_shares", quantity=_format_quantity(value))
+    quantity = _format_quantity(value)
+    key = "quantity_share" if quantity == "1" else "quantity_shares"
+    return translator(key, quantity=quantity)
 
 
 def _parse_detail_kwargs(text: str) -> dict[str, str]:
@@ -448,13 +467,39 @@ def _format_order_lines(
     for order in submitted:
         side = str(order.get("side") or "").lower()
         symbol = str(order.get("symbol") or "").upper()
-        side_key = "buy" if side == "buy" else "sell"
-        mode_key = "dry_run" if dry_run_only else "submitted"
+        raw_payload = dict(order.get("raw_payload") or {})
+        order_type = str(order.get("order_type") or raw_payload.get("price_type") or "limit").lower()
+        if order_type not in {"limit", "market"}:
+            order_type = "limit"
+        price = _format_price(order.get("limit_price") or raw_payload.get("limit_price") or raw_payload.get("price"))
+        price_suffix = translator("order_price_suffix", price=price) if price else ""
+        side_key = "side_buy" if side == "buy" else "side_sell"
+        order_type_key = "order_type_limit" if order_type == "limit" else "order_type_market"
+        quantity = _format_shares(order.get("quantity"), translator=translator)
+        if dry_run_only:
+            lines.append(
+                translator(
+                    "dry_run_order",
+                    order_type=translator(order_type_key),
+                    side=translator(side_key),
+                    symbol=symbol,
+                    quantity=quantity,
+                    price=price_suffix,
+                )
+            )
+            continue
+        order_id = str(order.get("broker_order_id") or raw_payload.get("order_id") or "").strip()
+        order_id_suffix = translator("order_id_suffix", order_id=order_id) if order_id else ""
         lines.append(
             translator(
-                f"{mode_key}_{side_key}_order",
+                "submitted_order",
+                icon="📈" if side == "buy" else "📉",
+                order_type=translator(order_type_key),
+                side=translator(side_key),
                 symbol=symbol,
-                quantity=_format_shares(order.get("quantity"), translator=translator),
+                quantity=quantity,
+                price=price_suffix,
+                order_id=order_id_suffix,
             )
         )
     return lines
@@ -515,8 +560,10 @@ def render_cycle_summary(result: Mapping[str, Any], *, lang: str = "en") -> str:
     lines.append(SEPARATOR)
     lines.extend(target_diff_lines)
     if submitted:
+        lines.append(translator("order_logs_title"))
         lines.extend(_format_order_lines(submitted, dry_run_only=dry_run_only, translator=translator))
     elif skipped and has_rebalance_attempt:
+        lines.append(translator("order_logs_title"))
         reason = _format_skipped_reason(skipped, translator=translator)
         lines.append(translator("no_order_submitted", reason=reason))
     else: