fix finance balance delta recharges

backend/app/database.py

@@ -41,7 +41,7 @@ def get_db():
 def init_db():
     """Create all tables."""
     # import models so SQLAlchemy registers them
-    from app.models import admin_user, upstream, snapshot, webhook_config, notification_log, custom_page, website, revoked_token, upstream_key, external_api_log, finance_daily_summary  # noqa: F401
+    from app.models import admin_user, upstream, snapshot, webhook_config, notification_log, custom_page, website, revoked_token, upstream_key, external_api_log, finance_daily_summary, upstream_recharge_event  # noqa: F401
     Base.metadata.create_all(bind=engine)
     _ensure_indexes()
     _migrate_custom_pages()
@@ -60,6 +60,8 @@ _NEW_INDEXES = [
     "CREATE INDEX IF NOT EXISTS ix_key_upstream_name ON upstream_generated_keys(upstream_id, key_name)",
     "CREATE INDEX IF NOT EXISTS ix_external_api_logs_created_at_desc ON external_api_logs(created_at DESC)",
     "CREATE INDEX IF NOT EXISTS ix_external_api_logs_direction_created ON external_api_logs(direction, created_at DESC)",
+    "CREATE INDEX IF NOT EXISTS ix_balance_snap_upstream_captured ON upstream_balance_snapshots(upstream_id, captured_at DESC)",
+    "CREATE INDEX IF NOT EXISTS ix_recharge_upstream_date ON upstream_recharge_events(upstream_id, recharge_date)",
 ]
 
 
@@ -132,6 +134,8 @@ def _migrate_upstreams():
             conn.execute(text("ALTER TABLE upstreams ADD COLUMN balance_alert_threshold FLOAT"))
         if "balance_alert_notified" not in columns:
             conn.execute(text("ALTER TABLE upstreams ADD COLUMN balance_alert_notified BOOLEAN NOT NULL DEFAULT 0"))
+        if "finance_cost_mode" not in columns:
+            conn.execute(text("ALTER TABLE upstreams ADD COLUMN finance_cost_mode VARCHAR(32) NOT NULL DEFAULT 'usage_stats'"))
 
 
 def _migrate_upstream_generated_keys():

backend/app/models/snapshot.py

@@ -1,5 +1,5 @@
 from datetime import datetime, timezone
-from sqlalchemy import Index, Integer, Text, DateTime, ForeignKey, text
+from sqlalchemy import Float, Index, Integer, Text, DateTime, ForeignKey, text
 from sqlalchemy.orm import mapped_column, Mapped
 from app.database import Base
 
@@ -15,3 +15,16 @@ class UpstreamRateSnapshot(Base):
     __table_args__ = (
         Index("ix_snapshot_upstream_captured", "upstream_id", text("captured_at DESC")),
     )
+
+
+class UpstreamBalanceSnapshot(Base):
+    __tablename__ = "upstream_balance_snapshots"
+
+    id: Mapped[int] = mapped_column(Integer, primary_key=True, index=True)
+    upstream_id: Mapped[int] = mapped_column(Integer, ForeignKey("upstreams.id", ondelete="CASCADE"), index=True)
+    balance: Mapped[float] = mapped_column(Float, nullable=False)
+    captured_at: Mapped[datetime] = mapped_column(DateTime, default=lambda: datetime.now(timezone.utc), index=True)
+
+    __table_args__ = (
+        Index("ix_balance_snap_upstream_captured", "upstream_id", text("captured_at DESC")),
+    )

backend/app/models/upstream.py

@@ -35,6 +35,8 @@ class Upstream(Base):
     # Balance alert
     balance_alert_threshold: Mapped[Optional[float]] = mapped_column(Float, nullable=True)
     balance_alert_notified: Mapped[bool] = mapped_column(Boolean, default=False)
+    # Finance cost mode: "usage_stats" (default) or "balance_delta"
+    finance_cost_mode: Mapped[str] = mapped_column(String(32), default="usage_stats")
     created_at: Mapped[datetime] = mapped_column(DateTime, default=lambda: datetime.now(timezone.utc))
     updated_at: Mapped[datetime] = mapped_column(
         DateTime, default=lambda: datetime.now(timezone.utc), onupdate=lambda: datetime.now(timezone.utc)

backend/app/models/upstream_recharge_event.py

@@ -0,0 +1,24 @@
+from datetime import date, datetime, timezone
+
+from sqlalchemy import Date, DateTime, Float, ForeignKey, Index, Integer, String, Text
+from sqlalchemy.orm import Mapped, mapped_column
+
+from app.database import Base
+
+
+class UpstreamRechargeEvent(Base):
+    __tablename__ = "upstream_recharge_events"
+
+    id: Mapped[int] = mapped_column(Integer, primary_key=True, index=True)
+    upstream_id: Mapped[int] = mapped_column(Integer, ForeignKey("upstreams.id", ondelete="CASCADE"), index=True)
+    recharge_date: Mapped[date] = mapped_column(Date, nullable=False)
+    amount: Mapped[float] = mapped_column(Float, nullable=False)
+    note: Mapped[str] = mapped_column(Text, default="")
+    created_at: Mapped[datetime] = mapped_column(DateTime, default=lambda: datetime.now(timezone.utc))
+    updated_at: Mapped[datetime] = mapped_column(
+        DateTime, default=lambda: datetime.now(timezone.utc), onupdate=lambda: datetime.now(timezone.utc)
+    )
+
+    __table_args__ = (
+        Index("ix_recharge_upstream_date", "upstream_id", "recharge_date"),
+    )

backend/app/routers/upstreams.py

@@ -4,18 +4,19 @@ from __future__ import annotations
 import json
 import logging
 import re
-from datetime import datetime, timezone
+from datetime import date as date_type, datetime, timezone
 from typing import Any, List
 
 logger = logging.getLogger(__name__)
 
-from fastapi import APIRouter, Depends, HTTPException, status
+from fastapi import APIRouter, Depends, HTTPException, Query as QueryParam, status
 from sqlalchemy.orm import Session
 
 from app.database import SessionLocal, get_db
 from app.models.admin_user import AdminUser
 from app.models.upstream import Upstream
 from app.models.upstream_key import UpstreamGeneratedKey
+from app.models.upstream_recharge_event import UpstreamRechargeEvent
 from app.models.snapshot import UpstreamRateSnapshot
 from app.schemas.upstream import (
     GenerateKeysByGroupsRequest,
@@ -23,10 +24,12 @@ from app.schemas.upstream import (
     GeneratedUpstreamKeyResponse,
     UpstreamCreate, UpstreamUpdate, UpstreamResponse, SnapshotResponse, TestResult,
     UpstreamBatchActionItem, UpstreamBatchActionSummary, UpstreamBatchActionResponse,
+    UpstreamRechargeCreate, UpstreamRechargeUpdate, UpstreamRechargeResponse,
 )
 from app.services.upstream_client import UpstreamClient, UpstreamError, _PendingKeyError, build_snapshot, build_new_api_token_name, mask_secret, _extract_key_value
 from app.services.auth_config import MASK, mask_auth_config, normalize_auth_config
-from app.services.snapshot_service import diff_snapshots
+from app.services.finance_service import today_shanghai
+from app.services.snapshot_service import diff_snapshots, write_balance_snapshot
 from app.services import scheduler as sched_svc
 from app.services import webhook_service
 from app.services import website_sync
@@ -138,11 +141,29 @@ def _to_response(u: Upstream) -> UpstreamResponse:
         balance_response_path=u.balance_response_path or "",
         balance_divisor=u.balance_divisor or 1.0,
         balance_alert_threshold=u.balance_alert_threshold,
+        finance_cost_mode=u.finance_cost_mode or "usage_stats",
         created_at=u.created_at,
         updated_at=u.updated_at,
     )
 
 
+def _recharge_response(row: UpstreamRechargeEvent) -> UpstreamRechargeResponse:
+    return UpstreamRechargeResponse(
+        id=row.id,
+        upstream_id=row.upstream_id,
+        date=row.recharge_date,
+        amount=row.amount,
+        note=row.note or "",
+        created_at=row.created_at,
+        updated_at=row.updated_at,
+    )
+
+
+def _validate_recharge_date(value: date_type) -> None:
+    if value > today_shanghai():
+        raise HTTPException(422, "date 不允许是未来日期")
+
+
 @router.get("", response_model=List[UpstreamResponse])
 def list_upstreams(db: Session = Depends(get_db), _=Depends(get_current_user)):
     return [_to_response(u) for u in db.query(Upstream).order_by(Upstream.id).all()]
@@ -694,6 +715,7 @@ def _test_upstream_core(db: Session, u: Upstream) -> UpstreamBatchActionItem:
                 raw_balance = client.get_balance(u.balance_endpoint, u.balance_response_path)
                 if raw_balance is not None:
                     u.balance = raw_balance / (u.balance_divisor or 1.0)
+                    write_balance_snapshot(db, u.id, u.balance)
                 u.balance_updated_at = datetime.now(timezone.utc) if raw_balance is not None else None
             except Exception as exc:
                 logger.warning("test-all: upstream %s balance failed: %s", u.name, exc)
@@ -738,6 +760,7 @@ def _check_now_core(db: Session, u: Upstream) -> tuple[str, bool]:
                 raw_balance = client.get_balance(u.balance_endpoint, u.balance_response_path)
                 if raw_balance is not None:
                     u.balance = raw_balance / (u.balance_divisor or 1.0)
+                    write_balance_snapshot(db, u.id, u.balance)
                 u.balance_updated_at = datetime.now(timezone.utc) if raw_balance is not None else None
             except Exception as exc:
                 logger.warning("check-now: upstream %s balance failed: %s", u.name, exc)
@@ -919,6 +942,7 @@ def create_upstream(
         balance_response_path=body.balance_response_path,
         balance_divisor=body.balance_divisor,
         balance_alert_threshold=body.balance_alert_threshold,
+        finance_cost_mode=body.finance_cost_mode,
     )
     db.add(u)
     db.commit()
@@ -935,6 +959,99 @@ def get_upstream(uid: int, db: Session = Depends(get_db), _=Depends(get_current_
     return _to_response(u)
 
 
+@router.get("/{uid}/recharges", response_model=List[UpstreamRechargeResponse])
+def list_recharges(
+    uid: int,
+    date: date_type | None = QueryParam(None),
+    db: Session = Depends(get_db),
+    _=Depends(get_current_user),
+):
+    u = db.query(Upstream).filter(Upstream.id == uid).first()
+    if not u:
+        raise HTTPException(404, "upstream not found")
+    query = db.query(UpstreamRechargeEvent).filter(UpstreamRechargeEvent.upstream_id == uid)
+    if date is not None:
+        query = query.filter(UpstreamRechargeEvent.recharge_date == date)
+    rows = query.order_by(UpstreamRechargeEvent.recharge_date.desc(), UpstreamRechargeEvent.id.desc()).all()
+    return [_recharge_response(row) for row in rows]
+
+
+@router.post("/{uid}/recharges", response_model=UpstreamRechargeResponse, status_code=201)
+def create_recharge(
+    uid: int,
+    body: UpstreamRechargeCreate,
+    db: Session = Depends(get_db),
+    _=Depends(get_current_user),
+):
+    u = db.query(Upstream).filter(Upstream.id == uid).first()
+    if not u:
+        raise HTTPException(404, "upstream not found")
+    _validate_recharge_date(body.date)
+    row = UpstreamRechargeEvent(
+        upstream_id=uid,
+        recharge_date=body.date,
+        amount=body.amount,
+        note=body.note or "",
+    )
+    db.add(row)
+    db.commit()
+    db.refresh(row)
+    return _recharge_response(row)
+
+
+@router.put("/{uid}/recharges/{rid}", response_model=UpstreamRechargeResponse)
+def update_recharge(
+    uid: int,
+    rid: int,
+    body: UpstreamRechargeUpdate,
+    db: Session = Depends(get_db),
+    _=Depends(get_current_user),
+):
+    u = db.query(Upstream).filter(Upstream.id == uid).first()
+    if not u:
+        raise HTTPException(404, "upstream not found")
+    row = (
+        db.query(UpstreamRechargeEvent)
+        .filter(UpstreamRechargeEvent.id == rid, UpstreamRechargeEvent.upstream_id == uid)
+        .first()
+    )
+    if not row:
+        raise HTTPException(404, "recharge event not found")
+    data = body.model_dump(exclude_unset=True)
+    if "date" in data:
+        _validate_recharge_date(data["date"])
+        row.recharge_date = data["date"]
+    if "amount" in data:
+        row.amount = data["amount"]
+    if "note" in data:
+        row.note = data["note"] or ""
+    row.updated_at = datetime.now(timezone.utc)
+    db.commit()
+    db.refresh(row)
+    return _recharge_response(row)
+
+
+@router.delete("/{uid}/recharges/{rid}", status_code=204)
+def delete_recharge(
+    uid: int,
+    rid: int,
+    db: Session = Depends(get_db),
+    _=Depends(get_current_user),
+):
+    u = db.query(Upstream).filter(Upstream.id == uid).first()
+    if not u:
+        raise HTTPException(404, "upstream not found")
+    row = (
+        db.query(UpstreamRechargeEvent)
+        .filter(UpstreamRechargeEvent.id == rid, UpstreamRechargeEvent.upstream_id == uid)
+        .first()
+    )
+    if not row:
+        raise HTTPException(404, "recharge event not found")
+    db.delete(row)
+    db.commit()
+
+
 @router.put("/{uid}", response_model=UpstreamResponse)
 def update_upstream(
     uid: int,
@@ -1029,8 +1146,6 @@ def latest_snapshot(uid: int, db: Session = Depends(get_db), _=Depends(get_curre
     )
 
 
-from fastapi import Query as QueryParam
-
 @router.get("/{uid}/snapshots", response_model=List[SnapshotResponse])
 def list_snapshots(
     uid: int,

backend/app/schemas/upstream.py

@@ -1,5 +1,5 @@
-from datetime import datetime
-from typing import Optional, Any
+from datetime import date, datetime
+from typing import Optional, Any, Literal
 from pydantic import BaseModel, Field
 
 
@@ -37,6 +37,7 @@ class UpstreamCreate(BaseModel):
     balance_response_path: str = ""
     balance_divisor: float = 1.0
     balance_alert_threshold: Optional[float] = None
+    finance_cost_mode: Literal["usage_stats", "balance_delta"] = "usage_stats"
 
 
 class UpstreamUpdate(BaseModel):
@@ -54,6 +55,7 @@ class UpstreamUpdate(BaseModel):
     balance_response_path: Optional[str] = None
     balance_divisor: Optional[float] = None
     balance_alert_threshold: Optional[float] = None
+    finance_cost_mode: Optional[Literal["usage_stats", "balance_delta"]] = None
 
 
 class UpstreamResponse(BaseModel):
@@ -77,12 +79,35 @@ class UpstreamResponse(BaseModel):
     balance_response_path: str = ""
     balance_divisor: float = 1.0
     balance_alert_threshold: Optional[float] = None
+    finance_cost_mode: Literal["usage_stats", "balance_delta"] = "usage_stats"
     created_at: datetime
     updated_at: datetime
 
     model_config = {"from_attributes": True}
 
 
+class UpstreamRechargeCreate(BaseModel):
+    date: date
+    amount: float = Field(gt=0)
+    note: str = ""
+
+
+class UpstreamRechargeUpdate(BaseModel):
+    date: Optional[date] = None
+    amount: Optional[float] = Field(default=None, gt=0)
+    note: Optional[str] = None
+
+
+class UpstreamRechargeResponse(BaseModel):
+    id: int
+    upstream_id: int
+    date: date
+    amount: float
+    note: str = ""
+    created_at: datetime
+    updated_at: datetime
+
+
 class SnapshotResponse(BaseModel):
     id: int
     upstream_id: int

backend/app/services/finance_service.py

@@ -21,10 +21,13 @@ from typing import Any
 
 import pytz
 
+from sqlalchemy import func
 from sqlalchemy.orm import Session
 
 from app.models.upstream import Upstream
 from app.models.website import Website
+from app.models.snapshot import UpstreamBalanceSnapshot
+from app.models.upstream_recharge_event import UpstreamRechargeEvent
 from app.services.upstream_client import UpstreamClient
 from app.services.website_client import Sub2ApiWebsiteClient, WebsiteError
 
@@ -81,6 +84,13 @@ def date_to_shanghai_timestamps(d: date) -> tuple[int, int]:
     return int(start.timestamp()), int(end.timestamp())
 
 
+def date_to_shanghai_utc_range(d: date) -> tuple[datetime, datetime]:
+    """Return UTC datetimes for [start, next_day_start) of a Shanghai calendar day."""
+    start = SHANGHAI_TZ.localize(datetime(d.year, d.month, d.day, 0, 0, 0))
+    end = start + timedelta(days=1)
+    return start.astimezone(timezone.utc), end.astimezone(timezone.utc)
+
+
 # ─────────────────────────────────────────────
 # Website revenue
 # ─────────────────────────────────────────────
@@ -229,6 +239,86 @@ def fetch_upstream_cost_new_api(upstream: Upstream, target_date: date) -> tuple[
         return 0.0, str(e)
 
 
+# ─────────────────────────────────────────────
+# Balance-delta cost (local balance snapshots)
+# ─────────────────────────────────────────────
+
+def fetch_upstream_cost_balance_delta(
+    upstream: Upstream,
+    target_date: date,
+    db: Session,
+) -> tuple[float, str | None]:
+    """Compute daily cost from local balance snapshots for balance_delta upstreams.
+
+    Algorithm:
+      1. Find the most recent snapshot BEFORE the start of target_date (baseline).
+      2. Find snapshots DURING target_date and use the last one as ending balance.
+      3. Sum manual recharge events for target_date.
+      4. cost = baseline + recharge_total - ending. Negative values fail.
+    """
+    start_dt, end_dt = date_to_shanghai_utc_range(target_date)
+
+    # Baseline: latest snapshot before start of target date
+    baseline = (
+        db.query(UpstreamBalanceSnapshot)
+        .filter(
+            UpstreamBalanceSnapshot.upstream_id == upstream.id,
+            UpstreamBalanceSnapshot.captured_at < start_dt,
+        )
+        .order_by(UpstreamBalanceSnapshot.captured_at.desc())
+        .first()
+    )
+
+    # Intra-day snapshots during target date
+    intra_day = (
+        db.query(UpstreamBalanceSnapshot)
+        .filter(
+            UpstreamBalanceSnapshot.upstream_id == upstream.id,
+            UpstreamBalanceSnapshot.captured_at >= start_dt,
+            UpstreamBalanceSnapshot.captured_at < end_dt,
+        )
+        .order_by(UpstreamBalanceSnapshot.captured_at.asc())
+        .all()
+    )
+
+    if baseline is None:
+        return 0.0, "缺少目标日期前的余额基线样本，无法进行差分统计"
+
+    if not intra_day:
+        return 0.0, "目标日期内无余额样本，无法进行差分统计"
+
+    recharge_total = (
+        db.query(func.coalesce(func.sum(UpstreamRechargeEvent.amount), 0.0))
+        .filter(
+            UpstreamRechargeEvent.upstream_id == upstream.id,
+            UpstreamRechargeEvent.recharge_date == target_date,
+        )
+        .scalar()
+    )
+    recharge_total = float(recharge_total or 0.0)
+
+    max_balance_rise = 0.0
+    prev = float(baseline.balance)
+    for snap in intra_day:
+        current = float(snap.balance)
+        if current > prev:
+            max_balance_rise = max(max_balance_rise, current - prev)
+        prev = current
+
+    if max_balance_rise > 0 and recharge_total <= 0:
+        return 0.0, "检测到余额上涨，但当天无充值记录；请补录充值后重新对账"
+
+    if max_balance_rise > 0 and recharge_total + 1e-9 < max_balance_rise:
+        return 0.0, "当天充值总额小于观察到的最大余额上涨，充值金额可能漏填"
+
+    ending = float(intra_day[-1].balance)
+    total = float(baseline.balance) + recharge_total - ending
+    if total < -1e-9:
+        return 0.0, "余额差分计算为负数，请检查余额样本或补录充值记录"
+
+    return round(max(total, 0.0), 6), None
+
+
 # ─────────────────────────────────────────────
 # Orchestration
 # ─────────────────────────────────────────────
@@ -237,6 +327,7 @@ def get_daily_summary(
     websites: list[Website],
     upstreams: list[Upstream],
     target_date: date,
+    db: Session | None = None,
 ) -> dict[str, Any]:
     """Compute daily revenue vs cost summary.
 
@@ -264,16 +355,24 @@ def get_daily_summary(
             total_revenue += amount
 
     for u in upstreams:
-        utype = classify_upstream(u)
-        if utype == "sub2api":
-            amount, err = fetch_upstream_cost_sub2api(u, target_date)
-        elif utype == "new_api":
-            amount, err = fetch_upstream_cost_new_api(u, target_date)
+        cost_mode = getattr(u, "finance_cost_mode", "usage_stats")
+        if cost_mode == "balance_delta":
+            if db is None:
+                amount, err = 0.0, "余额差分统计需要数据库连接"
+            else:
+                amount, err = fetch_upstream_cost_balance_delta(u, target_date, db)
+            utype = classify_upstream(u)
         else:
-            amount, err = 0.0, (
-                f"未知上游类型（auth_type={u.auth_type}, api_prefix={u.api_prefix}），"
-                "无法统计消费。请检查上游配置。"
-            )
+            utype = classify_upstream(u)
+            if utype == "sub2api":
+                amount, err = fetch_upstream_cost_sub2api(u, target_date)
+            elif utype == "new_api":
+                amount, err = fetch_upstream_cost_new_api(u, target_date)
+            else:
+                amount, err = 0.0, (
+                    f"未知上游类型（auth_type={u.auth_type}, api_prefix={u.api_prefix}），"
+                    "无法统计消费。请检查上游配置。"
+                )
 
         item = {
             "id": u.id,
@@ -339,7 +438,7 @@ def compute_daily_summary(
     """Compute a fresh daily summary (no DB write). Used for compare."""
     websites = db.query(Website).filter(Website.enabled == True).all()
     upstreams = db.query(Upstream).filter(Upstream.enabled == True).all()
-    return get_daily_summary(websites, upstreams, target_date)
+    return get_daily_summary(websites, upstreams, target_date, db=db)
 
 
 def _load_summary(row: Any) -> dict[str, Any]:

backend/app/services/scheduler.py

@@ -16,7 +16,7 @@ from app.models.upstream import Upstream
 from app.models.snapshot import UpstreamRateSnapshot
 from app.services.auth_config import normalize_auth_config
 from app.services.upstream_client import UpstreamClient, build_snapshot
-from app.services.snapshot_service import diff_snapshots, prune_snapshots
+from app.services.snapshot_service import diff_snapshots, prune_snapshots, write_balance_snapshot
 from app.services import webhook_service
 from app.services import website_sync
 from app.config import get_settings
@@ -102,6 +102,7 @@ def _check_upstream(upstream_id: int) -> None:
                 if balance is not None:
                     upstream.balance = balance
                     upstream.balance_updated_at = datetime.now(timezone.utc)
+                    write_balance_snapshot(db, upstream.id, balance)
                     # ── 余额告警阈值检查 ──
                     threshold = upstream.balance_alert_threshold
                     if threshold is not None and threshold > 0:
@@ -376,6 +377,16 @@ def start_scheduler() -> None:
             max_instances=1,
             misfire_grace_time=3600,
         )
+        # Daily balance snapshot capture at 23:59 Asia/Shanghai
+        _scheduler.add_job(
+            _capture_balance_snapshots,
+            trigger=CronTrigger(timezone="Asia/Shanghai", hour=23, minute=59),
+            id="balance_snapshot_daily",
+            replace_existing=True,
+            coalesce=True,
+            max_instances=1,
+            misfire_grace_time=600,
+        )
         logger.info("scheduler started with %d upstream job(s)", len(upstreams))
     finally:
         db.close()
@@ -416,6 +427,52 @@ def _compute_finance_daily_summary() -> None:
         db.close()
 
 
+def _capture_balance_snapshots() -> None:
+    """Fetch balance for all balance_delta-mode upstreams and write a snapshot.
+
+    Runs daily at 23:59 Asia/Shanghai to capture the end-of-day boundary balance.
+    Only writes a snapshot if balance fetch succeeds. Does NOT update the
+    upstream.balance field (that's the regular check's job).
+    """
+    db = SessionLocal()
+    try:
+        upstreams = db.query(Upstream).filter(
+            Upstream.enabled == True,
+            Upstream.finance_cost_mode == "balance_delta",
+            Upstream.balance_endpoint != "",
+            Upstream.balance_response_path != "",
+        ).all()
+        for u in upstreams:
+            try:
+                auth_config = json.loads(u.auth_config_json or "{}")
+                with UpstreamClient(
+                    base_url=u.base_url,
+                    api_prefix=u.api_prefix,
+                    auth_type=u.auth_type,
+                    auth_config=auth_config,
+                    timeout=float(u.timeout_seconds),
+                    on_auth_config_update=lambda updated: _persist_auth_config_update(u, updated),
+                    target_id=u.id,
+                    target_name=u.name,
+                ) as client:
+                    client.ensure_authenticated()
+                    raw_balance = client.get_balance(u.balance_endpoint, u.balance_response_path)
+                    if raw_balance is not None:
+                        balance = raw_balance / (u.balance_divisor or 1.0)
+                        write_balance_snapshot(db, u.id, balance)
+                        db.commit()
+                        logger.info("balance boundary snapshot for upstream %s: %.4f", u.name, balance)
+                    else:
+                        logger.warning("balance boundary snapshot: upstream %s returned None", u.name)
+            except Exception as exc:
+                db.rollback()
+                logger.warning("balance boundary snapshot failed for upstream %s: %s", u.name, exc)
+    except Exception:
+        logger.exception("failed to capture balance boundary snapshots")
+    finally:
+        db.close()
+
+
 def stop_scheduler() -> None:
     if _scheduler.running:
         _scheduler.shutdown(wait=True)

backend/app/services/snapshot_service.py

@@ -1,9 +1,10 @@
-"""Snapshot diff logic."""
+"""Snapshot diff logic and balance snapshot writes."""
+from datetime import datetime, timezone
 from typing import Any, Optional
 
 from sqlalchemy.orm import Session
 
-from app.models.snapshot import UpstreamRateSnapshot
+from app.models.snapshot import UpstreamBalanceSnapshot, UpstreamRateSnapshot
 
 
 def diff_snapshots(
@@ -43,6 +44,16 @@ def diff_snapshots(
     return changes
 
 
+def write_balance_snapshot(db: Session, upstream_id: int, balance: float) -> None:
+    """Write a balance snapshot row. Call only on successful balance fetch."""
+    row = UpstreamBalanceSnapshot(
+        upstream_id=upstream_id,
+        balance=balance,
+        captured_at=datetime.now(timezone.utc),
+    )
+    db.add(row)
+
+
 def prune_snapshots(db: Session, upstream_id: int, keep: int) -> None:
     if keep <= 0:
         return