fix: harden advisor market history dual-source sync

app/service/product_history_sync_service.py

@@ -21,6 +21,8 @@ NavHistoryLoader = Callable[[str, str, str], list[dict[str, str]]]
 class ProductHistorySyncResult:
     product_count: int
     observation_count: int
+    failed_product_count: int = 0
+    turnover_observation_count: int = 0
 
 
 class ProductHistorySyncService:
@@ -33,7 +35,10 @@ class ProductHistorySyncService:
         *,
         session_factory: Callable[[], Any] = SessionFactory,
         loader: NavHistoryLoader | None = None,
-        concurrency: int = 4,
+        # Public fund endpoints may reset connections when NAV and K-line requests
+        # for many products run concurrently. Keep the default conservative; callers
+        # can opt into a higher limit only when their provider permits it.
+        concurrency: int = 1,
     ) -> None:
         self.session_factory = session_factory
         self.loader = loader or self._default_loader
@@ -79,6 +84,8 @@ class ProductHistorySyncService:
         )
         now = datetime.now(UTC).replace(tzinfo=None)
         observations = 0
+        failed_products = 0
+        turnover_observations = 0
         for product_id, rows in fetched:
             payload = []
             for row in rows:
@@ -86,6 +93,7 @@ class ProductHistorySyncService:
                 if parsed is not None:
                     payload.append(parsed)
             if not payload:
+                failed_products += 1
                 continue
             async with self.session_factory() as session, session.begin():
                 upsert_statement = insert(AdvisorProductPriceHistory).values(payload)
@@ -97,7 +105,12 @@ class ProductHistorySyncService:
                     updated_at=upsert_statement.inserted.updated_at,
                 ))
             observations += len(payload)
-        return ProductHistorySyncResult(len(products), observations)
+            turnover_observations += sum(
+                row["turnover_amount"] is not None for row in payload
+            )
+        return ProductHistorySyncResult(
+            len(products), observations, failed_products, turnover_observations
+        )
 
     @classmethod
     def _row(
@@ -125,7 +138,7 @@ class ProductHistorySyncService:
             "price_kind": "fund_nav",
             "close_price": close_price,
             "turnover_amount": parsed_turnover,
-            "source": cls.SOURCE,
+            "source": str(raw.get("source") or cls.SOURCE),
             "source_updated_at": now,
             "created_at": now,
             "updated_at": now,