perf(pricebot): 透传复用共享 httpx 单例,免每请求重建 SSL 上下文 + 绕过进程代理

coupon/compare 透传原先每请求 async with httpx.AsyncClient(...) 新建:
每次重建 SSL 上下文(加载 certifi CA)实测 ~1s+,而 pricebot 是纯 HTTP
透传根本用不到 TLS;且 trust_env 默认 True 会把 http://localhost:8000
经进程代理(Clash)再绕几秒。

抽 app/core/pricebot_client.py 共享单例:trust_env=False 直连,lifespan
启动预热(SSL 一次性成本付在启动)、关停 aclose;timeout 下放到 .post()
保留 coupon 30s / compare 60s 差异。keep-alive 复用 TCP,每帧降到个位数 ms。

app/api/v1/ad.py

@@ -411,6 +411,7 @@ def feed_reward(payload: FeedRewardIn, user: CurrentUser, db: DbSession) -> Feed
         app_env=payload.app_env,
         our_code_id=payload.our_code_id,
         aborted=payload.aborted,
+        display_coin=payload.display_coin,
     )
     logger.info(
         "feed ad reward user_id=%d event=%s status=%s units=%d coin=%d",

app/api/v1/compare.py

@@ -26,6 +26,7 @@ import httpx
 from fastapi import APIRouter, HTTPException, Request, status
 
 from app.core.config import settings
+from app.core.pricebot_client import get_pricebot_client
 from app.core.pricebot_router import pick_pricebot
 
 logger = logging.getLogger("shagua.compare")
@@ -65,10 +66,10 @@ async def _passthrough(request: Request, upstream_path: str) -> dict[str, Any]:
     )
 
     try:
-        async with httpx.AsyncClient(timeout=timeout) as client:
-            resp = await client.post(
-                url, content=raw, headers={"Content-Type": "application/json"}
-            )
+        client = get_pricebot_client()
+        resp = await client.post(
+            url, content=raw, headers={"Content-Type": "application/json"}, timeout=timeout
+        )
     except httpx.RequestError as e:
         logger.error("[pricebot] request failed: %s", e)
         raise HTTPException(

app/api/v1/coupon.py

@@ -20,6 +20,7 @@ from fastapi.concurrency import run_in_threadpool
 
 from app.api.deps import CurrentUser, DbSession
 from app.core.config import settings
+from app.core.pricebot_client import get_pricebot_client
 from app.core.pricebot_router import pick_pricebot
 from app.db.session import SessionLocal
 from app.repositories import coupon_state as coupon_repo
@@ -141,10 +142,10 @@ async def coupon_step(
     )
 
     try:
-        async with httpx.AsyncClient(timeout=timeout) as client:
-            resp = await client.post(
-                url, content=raw, headers={"Content-Type": "application/json"}
-            )
+        client = get_pricebot_client()
+        resp = await client.post(
+            url, content=raw, headers={"Content-Type": "application/json"}, timeout=timeout
+        )
     except httpx.RequestError as e:
         logger.error("[pricebot] request failed: %s", e)
         raise HTTPException(

app/core/pricebot_client.py

@@ -0,0 +1,35 @@
+"""透传到 pricebot 的共享 httpx.AsyncClient 单例。
+
+为什么不能每请求新建(coupon.py / compare.py 老写法 async with httpx.AsyncClient(...)):
+  ① 每次构造都重建一套 SSL 上下文(httpx.create_ssl_context 加载 certifi CA),实测
+     ~1s+/次;而 pricebot 是纯 http 透传,根本用不到 TLS → 纯浪费,且每帧重交一次。
+  ② trust_env 默认 True 会读进程 HTTP_PROXY,把 http://localhost:8000 这条本地透传整个
+     塞进本机代理(如 Clash 7897),恒定再多几秒。
+单例:启动只建一次(SSL/连接池一次性),keep-alive 复用 TCP,每帧降到个位数 ms。
+trust_env=False:对齐 integrations/meituan.py 的既有约定,不被进程代理误导,直连 pricebot。
+"""
+from __future__ import annotations
+
+import httpx
+
+_client: httpx.AsyncClient | None = None
+
+
+def get_pricebot_client() -> httpx.AsyncClient:
+    """取透传单例。lifespan 启动会预热;未预热(如测试态)懒建兜底。
+
+    超时不在此固化(coupon 30s / compare 60s 不同),由调用点 client.post(timeout=...) 传。
+    懒建无 await,asyncio 单线程下不会有并发竞态。
+    """
+    global _client
+    if _client is None:
+        _client = httpx.AsyncClient(trust_env=False)
+    return _client
+
+
+async def aclose_pricebot_client() -> None:
+    """lifespan 关停时调,优雅关连接池。"""
+    global _client
+    if _client is not None:
+        await _client.aclose()
+        _client = None

app/main.py

@@ -49,6 +49,7 @@ from app.core.daily_exchange_worker import (
     stop_daily_exchange_worker,
 )
 from app.core.logging import setup_logging
+from app.core.pricebot_client import aclose_pricebot_client, get_pricebot_client
 from app.core.withdraw_reconcile_worker import (
     start_withdraw_reconcile_worker,
     stop_withdraw_reconcile_worker,
@@ -68,6 +69,7 @@ async def lifespan(_: FastAPI) -> AsyncIterator[None]:
         settings.APP_DEBUG,
         settings.DATABASE_URL.split("://", 1)[0],
     )
+    get_pricebot_client()  # 预热透传 client:把建 SSL 上下文的一次性成本付在启动,首个领券请求即热
     reconcile_task = start_withdraw_reconcile_worker()
     heartbeat_task = start_heartbeat_monitor()
     daily_exchange_task = start_daily_exchange_worker()
@@ -77,6 +79,7 @@ async def lifespan(_: FastAPI) -> AsyncIterator[None]:
         await stop_heartbeat_monitor(heartbeat_task)
         await stop_withdraw_reconcile_worker(reconcile_task)
         await stop_daily_exchange_worker(daily_exchange_task)
+        await aclose_pricebot_client()
         logger.info("shutting down")
 
 

app/repositories/ad_feed_reward.py

@@ -73,16 +73,19 @@ def grant_feed_reward(
     app_env: str | None = None,
     our_code_id: str | None = None,
     aborted: bool = False,
+    display_coin: int = 0,
 ) -> AdFeedRewardRecord:
     """**每条**信息流广告(客户端每条各上报一次)结算奖励。client_event_id 幂等,同号重试不重复发。
 
-    发奖规则:**一条广告 = 一个单次公式值**(rewards.calculate_ad_reward_coin),因子2(LT)按账号累计
-    **条**数递进;看满一份时长(unit_count>=1, 即 ≥10 秒)才发,**不逐份累加**。
+    发奖规则(所见即所得, 2026-06-27 用户拍板「显示多少给多少」):优先**直接发客户端小球显示的金币
+    display_coin**;防刷钳到本条「1 份满额」(eCPM 已钳 AD_ECPM_MAX_FEN, 因子2 按账号累计已发条数取档),
+    合法显示(实际因子2 × 进度 p ≤ 1 份)不被砍, 只挡伪造天价值。旧客户端不传 display_coin 时退回
+    「看满 10 秒发整份」(兼容不断币)。因子2(LT)由**客户端**按 granted 行 COUNT(拉自 /feed-reward/units)
+    算进 display_coin, 后端只记 granted 行让该计数自增, 不再服务端重算份值。
     - aborted=True(用户中途 ✕ 关闭这条):本条不发,记 status='closed_early'。
-    - 时长不足 10 秒(unit_count==0):记 status='too_short' 不发。
+    - display_coin 为 0 且时长不足一份:记 status='too_short' 不发(不计 LT / 当日上限)。
     - 命中当日条数上限:记 status='capped' 不发。
-    duration_seconds 是**这一条**的观看秒数。服务端两道硬闸防刷:时长钳到 FEED_MAX_DURATION_SECONDS、
-    eCPM 在 calculate_ad_reward_coin 内钳到 AD_ECPM_MAX_FEN;叠加每日 get_ad_daily_limit 条数上限。
+    duration_seconds 落库留痕(unit_count 字段), 旧端兼容路径据它判是否满 1 份。
     feed_scene:点位场景(comparison/coupon/welfare),仅归类落库,不参与计算。
     ad_type:广告形态(feed 信息流 / draw Draw 信息流),仅归类落库;**每日上限与因子2(LT)仍按本表
     全表 unit 累计(feed+draw 共享同一发奖池/上限),不按 ad_type 拆分**。
@@ -139,14 +142,26 @@ def grant_feed_reward(
         )
         return _commit_record(db, rec, client_event_id)
 
-    # 整场总时长不足 10 秒,凑不满一份 → 不发,记 too_short 留痕。
-    if unit_count == 0:
+    # 所见即所得(用户 2026-06-27「显示多少给多少」): 优先发**客户端小球显示**的金币 display_coin,
+    # 钳到本条「1 份满额」防刷(eCPM 已钳 AD_ECPM_MAX_FEN; 合法显示=因子2×p≤1份, 不会被砍)。
+    # 因子2(LT)按账号累计已发条数(granted 行 COUNT), 第 existing_ads+1 条。
+    existing_ads = granted_unit_total(db, user_id)
+    unit_cap = rewards.calculate_ad_reward_coin(ecpm, existing_ads + 1)
+    if display_coin > 0:
+        coin = min(display_coin, unit_cap)              # 新端: 所见即所得(直接发小球显示金币)
+    elif unit_count >= 1:
+        coin = unit_cap                                 # 旧端没传 display_coin: 退回「看满 1 份发整份」(兼容)
+    else:
+        coin = 0
+
+    # 显示金币为 0 且没满一份 → 不发, 记 too_short 留痕(不写 granted 行 → 不计 LT / 当日上限)。
+    if coin <= 0:
         rec = AdFeedRewardRecord(
             client_event_id=client_event_id,
             user_id=user_id,
             reward_date=today,
             duration_seconds=safe_duration,
-            unit_count=0,
+            unit_count=unit_count,
             ad_session_id=ad_session_id,
             ecpm_raw=ecpm,
             adn=adn,
@@ -161,15 +176,11 @@ def grant_feed_reward(
         )
         return _commit_record(db, rec, client_event_id)
 
-    # 一条广告 = 一个「单次公式值」(因子2 按账号累计**条**数, 即第 existing_ads+1 条);看满一份(unit_count>=1)即发,不逐份累加。
-    existing_ads = granted_unit_total(db, user_id)
-    coin = rewards.calculate_ad_reward_coin(ecpm, existing_ads + 1)
-    if coin > 0:
-        crud_wallet.grant_coins(
-            db, user_id, coin,
-            biz_type="feed_ad_reward", ref_id=client_event_id,
-            remark="信息流广告奖励",
-        )
+    crud_wallet.grant_coins(
+        db, user_id, coin,
+        biz_type="feed_ad_reward", ref_id=client_event_id,
+        remark="信息流广告奖励",
+    )
     rec = AdFeedRewardRecord(
         client_event_id=client_event_id,
         user_id=user_id,

app/schemas/ad.py

@@ -167,6 +167,11 @@ class FeedRewardIn(BaseModel):
     aborted: bool = Field(
         False, description="用户中途 ✕ 关闭广告(未走完比价):整场不发,记 closed_early"
     )
+    display_coin: int = Field(
+        0, ge=0,
+        description="客户端金币小球**本条显示**的金币(所见即所得):后端直接发这个数,钳到本条最大 1 份"
+        "满额防刷。缺省 0 = 旧客户端不传,退回服务端「看满 10 秒发整份」",
+    )
 
 
 class FeedRewardOut(BaseModel):