支持Cpa按Key来拼车

README.md

@@ -16,7 +16,7 @@
 - 平均恒定模型保留透明的累计比例对照；时变额度模型使用混合粒子滤波估计连续容量、整数显示规则和参与者归属，并同时给出概率区间与确定性边界。
 - 在参与者用户余额耗尽、建议余额变化、测算率变化或采集异常时发送邮件通知。
 - 支持 SMTP 和 Resend 邮件服务。
-- 提供额度总览、账号状态、观测记录、粒子轨迹、额度统计、登录审计和使用教程；CPA 统计按本地哈希后的 API Key 分组，并显示未知模型的未计价请求数，不创建参与者、余额或额度建议。
+- 提供额度总览、账号状态、观测记录、粒子轨迹、额度统计、登录审计和使用教程；CPA 支持参与者绑定多个 Key、独立或多账号额度池、个人权益与同车成员用量汇总，以及本人请求明细；历史请求可预览后认领，未计价模型和采集缺口显式保留。配置见 [CPA 拼车](docs/cpa-carpool.md)。
 - 管理员可为普通系统用户逐页配置只读访问权限，并精确选择其可见账号与参与者；账号范围同时约束额度总览、账号状态、观测、粒子轨迹和统计，所有页面写操作仍仅限管理员。系统设置页无需授权且不可关闭，普通用户在其中只会看到自己的 API Key 卡片。
 - 提供内置 OpenAPI 3.1 文档的业务 API；管理员全局 Bearer Key 拥有全部已开放权限并可一键设置建议余额，普通系统用户可在系统设置页生成绑定自身的 Key，其端点发现、OpenAPI 文档和业务响应实时继承可配置页面、账号与参与者权限，且不能执行管理写操作。
 - 记录服务端来源 IP，并可选记录浏览器通过 WebRTC 上报的辅助地址。

backend/monitor/access.py

@@ -2,7 +2,7 @@
 
 from collections.abc import Iterable
 
-from django.db.models import QuerySet
+from django.db.models import Q, QuerySet
 from rest_framework.permissions import BasePermission
 
 from .models import (
@@ -86,13 +86,15 @@ def visible_accounts_for(user, queryset: QuerySet | None = None) -> QuerySet:
     accounts = queryset if queryset is not None else MonitoredAccount.objects.all()
     if user.is_staff:
         return accounts
-    return accounts.filter(authorized_users=user)
+    return accounts.filter(authorized_users=user).filter(
+        Q(provider="sub2api") | Q(pool__allocations__participant__authorized_users=user)
+    ).distinct()
 
 
 def visible_account_ids(user) -> set[int] | None:
     if user.is_staff:
         return None
-    return set(user.visible_monitored_accounts.values_list("id", flat=True))
+    return set(visible_accounts_for(user).values_list("id", flat=True))
 
 
 def scope_participant_data(

backend/monitor/accounting/adaptive_range.py

@@ -50,14 +50,19 @@ class AdaptiveRangeOutput:
     direction: ExpansionDirection | None
     promotions: tuple[RangePromotion, ...]
     filter_config: ParticleFilterConfig
+    initial_range_usd: tuple[float, float]
 
 
 def _display_residuals(
     particle: ParticleFilterOutput,
     displayed_percent: np.ndarray,
 ) -> tuple[np.ndarray, np.ndarray]:
-    # 与研究实现一致：各主体后验中位数之和用于检查显示单元矛盾。
-    progress = particle.attributed_percent_hat.sum(axis=1)
+    # Compare absolute progress with the provider's absolute display. Baseline
+    # uncertainty is inferred by the filter, not forced to the displayed integer.
+    # Exact-zero baselines retain the validated sum-of-subject-medians behavior.
+    progress = (
+        particle.baseline_percent_hat + particle.attributed_percent_hat.sum(axis=1)
+    )
     upper = np.maximum(
         progress - np.minimum(displayed_percent + 1.0, 100.0),
         0.0,
@@ -193,27 +198,41 @@ def run_adaptive_range_filter(
         capacity_max_usd=capacity_profile.capacity_max_usd,
         initial_capacity_sd_usd=capacity_profile.initial_capacity_sd_usd,
     )
+    # A prior learned outside the base plan range must not be clipped back to
+    # its boundary on reconnect. Reuse the smallest supported range containing
+    # that prior; this is historical input, not a new dual-evidence promotion.
+    prior = base_config.initial_capacity_usd
+    if prior is not None and np.isfinite(prior) and prior > 0:
+        for target in capacity_profile.lower_stages_usd:
+            if prior >= base_config.capacity_min_usd:
+                break
+            base_config = replace(base_config, capacity_min_usd=target)
+        for target in capacity_profile.upper_stages_usd:
+            if prior <= base_config.capacity_max_usd:
+                break
+            base_config = replace(base_config, capacity_max_usd=target)
+    initial_range = (base_config.capacity_min_usd, base_config.capacity_max_usd)
     base_particle = run_particle_filter(
         model_input,
         seed=seed,
         config=base_config,
     )
     base_bounds = run_deterministic_bounds(
         model_input,
-        capacity_min_usd=capacity_profile.capacity_min_usd,
-        capacity_max_usd=capacity_profile.capacity_max_usd,
+        capacity_min_usd=base_config.capacity_min_usd,
+        capacity_max_usd=base_config.capacity_max_usd,
     )
     particle = _copy_particle(base_particle)
     bounds = _copy_bounds(base_bounds)
     row_count = len(model_input.times_hours)
     active_min = np.full(
         row_count,
-        capacity_profile.capacity_min_usd,
+        base_config.capacity_min_usd,
         dtype=float,
     )
     active_max = np.full(
         row_count,
-        capacity_profile.capacity_max_usd,
+        base_config.capacity_max_usd,
         dtype=float,
     )
     active_stage = np.zeros(row_count, dtype=int)
@@ -232,6 +251,7 @@ def run_adaptive_range_filter(
             direction=None,
             promotions=(),
             filter_config=base_config,
+            initial_range_usd=initial_range,
         )
 
     targets = (
@@ -240,8 +260,12 @@ def run_adaptive_range_filter(
         else capacity_profile.lower_stages_usd
     )
     current_particle = base_particle
-    current_min = capacity_profile.capacity_min_usd
-    current_max = capacity_profile.capacity_max_usd
+    current_min = base_config.capacity_min_usd
+    current_max = base_config.capacity_max_usd
+    targets = tuple(
+        t for t in targets
+        if (t > current_max if direction == "upper" else t < current_min)
+    )
     promotions: list[RangePromotion] = []
 
     for stage_number, target in enumerate(targets, start=1):
@@ -311,4 +335,5 @@ def run_adaptive_range_filter(
         direction=direction,
         promotions=tuple(promotions),
         filter_config=base_config,
+        initial_range_usd=initial_range,
     )

backend/monitor/accounting/contracts.py

@@ -6,7 +6,7 @@
 
 from ..models import Observation
 
-ALGORITHM_VERSION = "particle_filter_v9"
+ALGORITHM_VERSION = "particle_filter_v10"
 
 
 @dataclass

backend/monitor/accounting/dynamic_attribution.py

@@ -139,6 +139,7 @@ def _diagnostics(
             float(adaptive.capacity_min_usd[row]),
             float(adaptive.capacity_max_usd[row]),
         ],
+        "capacity_initial_range_usd": list(adaptive.initial_range_usd),
         "capacity_range_stage": int(adaptive.stage[row]),
         "capacity_range_direction": (
             adaptive.direction if adaptive.stage[row] > 0 else None
@@ -543,6 +544,17 @@ def replay_dynamic_segment(
                 snapshot.recommendation_applied = False
             snapshot.needs_manual_update = needs_update
             snapshot.reason = reason
+            if account_id < 0:
+                snapshot.current_balance_usd = None
+                snapshot.recommended_balance_usd = None
+                snapshot.recommended_balance_min_usd = None
+                snapshot.recommended_balance_max_usd = None
+                snapshot.deterministic_balance_min_usd = None
+                snapshot.deterministic_balance_max_usd = None
+                snapshot.balance_difference_usd = None
+                snapshot.needs_manual_update = False
+                snapshot.recommendation_applied = False
+                snapshot.reason = "CPA 本地估算权益" if snapshot.cpa_contract_known else "缺少历史份额，剩余权益未知"
         if snapshots:
             ParticipantSnapshot.objects.bulk_update(
                 snapshots,

backend/monitor/accounting/dynamic_contracts.py

@@ -55,6 +55,7 @@ class ParticleFilterOutput:
     capacity_hat_usd: np.ndarray
     capacity_lower_usd: np.ndarray
     capacity_upper_usd: np.ndarray
+    baseline_percent_hat: np.ndarray
     total_percent_hat: np.ndarray
     total_percent_lower: np.ndarray
     total_percent_upper: np.ndarray

backend/monitor/accounting/model_inputs.py

@@ -74,23 +74,25 @@ def build_dynamic_replay_input(
         user_ids.add(row.sub2api_user_id)
 
     participant_by_id = {}
+    source_by_participant = {}
     share_by_participant_id = {}
     snapshots_by_observation: list[dict[int, object]] = []
     for observation in observations:
         by_user = {}
         for snapshot in observation.participant_snapshots.all():
             participant = snapshot.participant
+            source_by_participant[participant.id] = participant.id if account_id < 0 else (participant.sub2api_user_id if participant.sub2api_user_id is not None else snapshot.source_sub2api_user_id)
             participant_by_id[participant.id] = participant
             share_by_participant_id[participant.id] = snapshot.share_percent
-            user_ids.add(participant.sub2api_user_id)
-            by_user[participant.sub2api_user_id] = snapshot
+            user_ids.add(source_by_participant[participant.id])
+            by_user[source_by_participant[participant.id]] = snapshot
         snapshots_by_observation.append(by_user)
 
     ordered_users = tuple(sorted(user_ids))
     subject_user_ids: tuple[int | None, ...] = (*ordered_users, RESIDUAL_SUBJECT)
     user_index = {user_id: index for index, user_id in enumerate(ordered_users)}
     participant_subject_indices = {
-        participant_id: user_index[participant.sub2api_user_id]
+        participant_id: user_index[source_by_participant[participant.id]]
         for participant_id, participant in participant_by_id.items()
     }
     rights = np.zeros(len(subject_user_ids), dtype=float)
@@ -114,7 +116,7 @@ def build_dynamic_replay_input(
             participant = participant_by_id.get(participant_id)
             if participant is not None:
                 baseline_by_user.setdefault(
-                    participant.sub2api_user_id,
+                    source_by_participant[participant.id],
                     baseline,
                 )
 

backend/monitor/accounting/particle_filter.py

@@ -155,6 +155,7 @@ def run_particle_filter(
     capacity_hat = np.zeros(observation_count)
     capacity_lower = np.zeros(observation_count)
     capacity_upper = np.zeros(observation_count)
+    baseline_hat = np.zeros(observation_count)
     total_hat = np.zeros(observation_count)
     total_lower = np.zeros(observation_count)
     total_upper = np.zeros(observation_count)
@@ -200,6 +201,9 @@ def record(
         upper_boundary_mass[index] = weights[
             capacity_particles >= cfg.capacity_max_usd - boundary_band
         ].sum()
+        baseline_hat[index] = _weighted_quantile(
+            baseline_hidden, weights, (0.5,),
+        )[0]
         total_particles = attributed_particles.sum(axis=1)
         remaining_percent = np.maximum(
             model_input.rights_percent[None, :] - attributed_particles,
@@ -384,6 +388,7 @@ def record(
         capacity_hat_usd=capacity_hat,
         capacity_lower_usd=capacity_lower,
         capacity_upper_usd=capacity_upper,
+        baseline_percent_hat=baseline_hat,
         total_percent_hat=total_hat,
         total_percent_lower=total_lower,
         total_percent_upper=total_upper,

backend/monitor/accounting/replay.py

@@ -88,7 +88,9 @@ def _replay_usage_samples(
             observed_at__in=[sample.observed_at for sample in samples],
         )
     }
+    historical_sources = {row.participant_id: row.source_sub2api_user_id for row in ParticipantSnapshot.objects.filter(observation__account_id=account_id).order_by("observation__observed_at", "id")}
     for sample in samples:
+        source_id = sample.participant_id if account_id < 0 else (sample.participant.sub2api_user_id if sample.participant.sub2api_user_id is not None else historical_sources.get(sample.participant_id))
         segment = None
         for candidate in segments:
             if candidate.first_observed_at <= sample.observed_at:
@@ -101,7 +103,7 @@ def _replay_usage_samples(
             continue
         sample.attribution_started_at = segment.started_at
         raw_cost = normalized_by_key.get(
-            (sample.participant.sub2api_user_id, sample.observed_at),
+            (source_id, sample.observed_at),
             sample.raw_selected_cost,
         )
         baseline = segment.participant_baselines.get(
@@ -115,7 +117,7 @@ def _replay_usage_samples(
         }:
             baseline = normalized_by_key.get(
                 (
-                    sample.participant.sub2api_user_id,
+                    source_id,
                     segment.first_observed_at,
                 ),
                 baseline,
@@ -125,7 +127,7 @@ def _replay_usage_samples(
             raw_cost
             - baseline
             + correction_prefix.user_between(
-                sample.participant.sub2api_user_id,
+                source_id,
                 segment.started_at,
                 sample.observed_at,
             ),
@@ -151,7 +153,7 @@ def _update_participant_latest(account_id: int) -> None:
         return
     snapshots = list(latest.participant_snapshots.all())
     participant_ids = [snapshot.participant_id for snapshot in snapshots]
-    account = MonitoredAccount.objects.filter(external_account_id=account_id).first()
+    account = MonitoredAccount.for_fact_key(account_id)
     memberships = (
         {
             item.participant_id: item
@@ -177,10 +179,14 @@ def _update_participant_latest(account_id: int) -> None:
             ["latest_selected_cost", "last_checked_at"],
         )
 
+    if account_id < 0:
+        return
+
     latest_balance_sample = ParticipantBalanceSample.objects.filter(
         participant_id=OuterRef("pk"),
     ).order_by("-captured_at", "-id")
     latest_snapshot = ParticipantSnapshot.objects.filter(
+        observation__account_id__gt=0,
         participant_id=OuterRef("pk"),
         observation__excluded_at__isnull=True,
     ).order_by("-observation__observed_at", "-id")
@@ -343,6 +349,9 @@ def rebuild_account(
             CPAAccountCollectionInterval.objects.filter(account=monitored_account)
             .order_by("connected_at", "id")
         )
+    if monitored_account is not None and monitored_account.provider == "cpa":
+        from ..cpa.participants import materialize_participants
+        materialize_participants(monitored_account, config)
     all_observations = list(
         Observation.objects.select_for_update()
         .select_related("sample_point", "manual_start_end")

backend/monitor/api_usage.py

@@ -208,7 +208,7 @@ def refresh_due_api_usage_snapshots(config: AppSettings) -> dict[str, int]:
         observation = latest_cycle_observation(account)
         if observation is None:
             continue
-        participants = Participant.objects.filter(enabled=True)
+        participants = Participant.objects.filter(enabled=True, sub2api_user_id__isnull=False)
         for participant in participants:
             if fresh_snapshot(
                 participant=participant,