feat(scoring): 适老画像口径 41+→纯50+ + 优质老年作者入池规则

改动1(口径统一纯50+):fifty_plus.py BAND_KEYS 只留 ('50-',)、band 标签 50+;
评分详情文案 41+→50+。50+子分(抖音30%权重)、scorecard.fifty_plus_components 全转纯50+。

改动2(优质老年作者入池):复用已建空表 content_agent_author_assets/_roles + 已接入的
result_source_lookup._build_author_assets;入池门换成 纯50+占比>30% 且 TGI>120(读
decision.scorecard.fifty_plus_components),填 elderly_ratio/elderly_tgi、打 high_50plus_profile 角色。
弃用遗留 age_50_plus_level 采样规则(死字段保留)。不新建表、不改 evaluator 权重/walk。

测试:fifty_plus 公式/douyin评分/walk helper 改纯50+并重算;新增 author 入池用例。512 passed。

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

content_agent/business_modules/content_discovery/fifty_plus.py

@@ -1,6 +1,6 @@
 """M9A:抖音 50+ 受众子分(作者粉丝画像 → 0-100)。
 
-口径(已拍板):band=41+(41-50 + 50-=50 岁及以上);
+口径(已拍板):band=50+(仅 50-=50 岁及以上,不含 41-50 中年);
 TGI_score=clamp(bandTGI/2,0,100)(TGI 200 满分);percent_score=clamp(band占比,0,100);
 50+子分=0.6·TGI_score+0.4·percent_score。TGI=画像 preference,占比=画像 percentage,同一接口。
 """
@@ -10,7 +10,7 @@ from __future__ import annotations
 from typing import Any
 
 SCHEMA_VERSION = "v4_fifty_plus.v1"
-BAND_KEYS = ("41-50", "50-")  # 50- = 50 岁及以上
+BAND_KEYS = ("50-",)  # 50- = 50 岁及以上(纯 50+,已弃用 41-50)
 
 
 def fifty_plus_score(fans_age_data: dict[str, Any] | None) -> dict[str, Any]:
@@ -31,7 +31,7 @@ def fifty_plus_score(fans_age_data: dict[str, Any] | None) -> dict[str, Any]:
         pairs.append((pct, tgi))
 
     if not pairs:
-        return {"schema_version": SCHEMA_VERSION, "status": "incomplete", "band": "41+"}
+        return {"schema_version": SCHEMA_VERSION, "status": "incomplete", "band": "50+"}
 
     band_percent = sum(pct for pct, _ in pairs)
     weight = band_percent if band_percent > 0 else float(len(pairs))
@@ -43,7 +43,7 @@ def fifty_plus_score(fans_age_data: dict[str, Any] | None) -> dict[str, Any]:
         "schema_version": SCHEMA_VERSION,
         "status": "ok",
         "score": score,
-        "band": "41+",
+        "band": "50+",
         "band_tgi": round(band_tgi, 2),
         "band_percent": round(band_percent, 2),
         "tgi_score": round(tgi_score, 2),

content_agent/business_modules/result_source_lookup.py

@@ -426,8 +426,9 @@ def _build_author_assets(
         sample_count = len(items)
         qualified_count = len(qualified_decisions)
         qualified_ratio = qualified_count / sample_count if sample_count else 0
-        age_level = _best_age_level(decisions)
-        if not _author_asset_eligible(sample_count, qualified_count, qualified_ratio, age_level):
+        # 优质老年作者入池门:纯 50+ 画像(来自 fifty_plus,仅抖音有)占比>30% 且 TGI>120。
+        elderly_ratio, elderly_tgi = _best_fifty_plus(decisions)
+        if not _author_asset_eligible(elderly_ratio, elderly_tgi):
             continue
 
         author_asset_id = _stable_id("author_asset", platform, author_id)
@@ -455,7 +456,8 @@ def _build_author_assets(
             "sample_count": sample_count,
             "qualified_content_count": qualified_count,
             "qualified_content_ratio": qualified_ratio,
-            "age_50_plus_level": age_level,
+            "elderly_ratio": elderly_ratio,
+            "elderly_tgi": elderly_tgi,
         }
         summary = {
             "author_asset_id": author_asset_id,
@@ -481,8 +483,8 @@ def _build_author_assets(
                 "source_type": source_type,
                 "validation_status": "rule_validated",
                 "eligible_as_source": 1,
-                "elderly_ratio": None,
-                "elderly_tgi": None,
+                "elderly_ratio": elderly_ratio,
+                "elderly_tgi": elderly_tgi,
                 "content_tags": tags,
                 "source_run_id": run_id,
                 "source_policy_run_id": policy_run_id,
@@ -499,7 +501,7 @@ def _build_author_assets(
                 "author_asset_id": author_asset_id,
                 "role": role,
                 "role_status": "active",
-                "role_reason_code": "p7_author_asset_eligible",
+                "role_reason_code": "elderly_50plus_profile_pass",
                 "assigned_by": "system",
                 "source_run_id": run_id,
                 "raw_payload": {"evidence_refs": evidence_refs},
@@ -509,20 +511,38 @@ def _build_author_assets(
     return summaries, asset_rows, role_rows
 
 
-def _author_asset_eligible(
-    sample_count: int,
-    qualified_count: int,
-    qualified_ratio: float,
-    age_level: str,
-) -> bool:
+def _author_asset_eligible(elderly_ratio: float | None, elderly_tgi: float | None) -> bool:
+    """优质老年作者入池门:纯 50+ 占比 > 30% 且 50+ TGI > 120(仅抖音有画像,天然只抖音入池)。"""
     return (
-        sample_count >= 9
-        and qualified_count >= 3
-        and qualified_ratio >= 1 / 3
-        and age_level in {"medium", "strong"}
+        elderly_ratio is not None
+        and elderly_tgi is not None
+        and elderly_ratio > 30
+        and elderly_tgi > 120
     )
 
 
+def _best_fifty_plus(decisions: list[dict[str, Any]]) -> tuple[float | None, float | None]:
+    """从作者各 decision 的 scorecard 取 50+ 画像(status=ok),按 50+占比取最强的一条。
+
+    返回 (band_percent=纯50+占比, band_tgi=纯50+TGI);无 ok 画像 → (None, None)。
+    """
+    best: tuple[float, float] | None = None
+    for decision in decisions:
+        scorecard = decision.get("scorecard")
+        if not isinstance(scorecard, dict) or scorecard.get("fifty_plus_status") != "ok":
+            continue
+        components = scorecard.get("fifty_plus_components")
+        if not isinstance(components, dict):
+            continue
+        pct = components.get("band_percent")
+        tgi = components.get("band_tgi")
+        if not isinstance(pct, (int, float)) or not isinstance(tgi, (int, float)):
+            continue
+        if best is None or pct > best[0]:
+            best = (float(pct), float(tgi))
+    return best if best is not None else (None, None)
+
+
 def _best_age_level(decisions: list[dict[str, Any]]) -> str:
     priority = {"strong": 3, "medium": 2, "weak": 1, "missing": 0}
     levels = [decision.get("age_50_plus_level", "missing") for decision in decisions]

content_agent/flow_ledger_service.py

@@ -1507,7 +1507,7 @@ def _score_items_from_values(
         # 50+老年 维度卡(仅抖音):status=ok 出真实分,否则出灰卡「未评估」。
         if fifty.get("status") == "ok":
             items.append(_score_item(
-                "50+老年受众", fifty.get("score"), "作者粉丝 41+ 画像(TGI × 占比)",
+                "50+老年受众", fifty.get("score"), "作者粉丝 50+ 画像(TGI × 占比)",
                 weight=weights.get("fifty"), group="main", status="ok",
             ))
         else:
@@ -1527,11 +1527,11 @@ def _score_items_from_values(
         # 50+老年怎么算:TGI 表现(60%)+ 占比表现(40%),取自 fifty_plus_components。
         comps = _record(fifty.get("components"))
         items.append(_score_item(
-            "TGI 表现", comps.get("tgi_score"), f"41+ 粉丝 TGI≈{_num_text(comps.get('band_tgi'))}(200 满分)",
+            "TGI 表现", comps.get("tgi_score"), f"50+ 粉丝 TGI≈{_num_text(comps.get('band_tgi'))}(200 满分)",
             weight=0.6, group="fifty",
         ))
         items.append(_score_item(
-            "占比表现", comps.get("percent_score"), f"41+ 粉丝占比≈{_num_text(comps.get('band_percent'))}%",
+            "占比表现", comps.get("percent_score"), f"50+ 粉丝占比≈{_num_text(comps.get('band_percent'))}%",
             weight=0.4, group="fifty",
         ))
     return [item for item in items if item]

tests/p6_walk_helpers.py

@@ -35,14 +35,13 @@ class FakeWalkPlatformClient:
         return [_platform_result(query, "7390000000000000301", "作者作品", [])]
 
     def fetch_account_fans_portrait(self, account_id: str) -> dict[str, Any]:
-        # M9:强 41+ 画像(像广场舞号),50+ 子分高 → 不改变 walk 既有 allow_walk/入池 行为。
+        # 强 50+ 画像(纯 50- 桶,占比60%/TGI200 → 50+子分高)→ 不改变 walk 既有 allow_walk/入池 行为。
         self.portrait_calls.append(account_id)
         return {
             "fans": {
                 "age": {
                     "data": {
-                        "41-50": {"percentage": "31.73%", "preference": "210.02"},
-                        "50-": {"percentage": "29.76%", "preference": "130.53"},
+                        "50-": {"percentage": "60%", "preference": "200"},
                     }
                 }
             }

tests/test_author_asset_pool.py

@@ -0,0 +1,77 @@
+"""优质老年作者入池:纯50+占比>30% 且 TGI>120 才收;填 elderly_ratio/tgi;非抖音/不达标不收。"""
+
+from content_agent.business_modules.result_source_lookup import _build_author_assets
+
+
+def _item(author_id, content_id, platform="douyin"):
+    return {
+        "platform": platform,
+        "platform_author_id": author_id,
+        "platform_content_id": content_id,
+        "content_discovery_id": f"cd_{content_id}",
+        "author_display_name": f"作者{author_id}",
+        "tags": ["历史"],
+    }
+
+
+def _decision(content_id, fifty_status=None, band_percent=None, band_tgi=None):
+    scorecard = {"total_score": 73, "query_relevance_score": 90, "platform_performance_score": 70}
+    if fifty_status is not None:
+        scorecard["fifty_plus_status"] = fifty_status
+        if fifty_status == "ok":
+            scorecard["fifty_plus_components"] = {"band": "50+", "band_percent": band_percent, "band_tgi": band_tgi}
+    return {
+        "decision_id": f"dec_{content_id}",
+        "decision_target_id": content_id,
+        "decision_action": "ADD_TO_CONTENT_POOL",
+        "scorecard": scorecard,
+    }
+
+
+def _run(items, decisions):
+    decision_by_target_id = {d["decision_target_id"]: d for d in decisions}
+    paths = {it["platform_content_id"]: [] for it in items}
+    return _build_author_assets("run", "policy_run", items, decision_by_target_id, paths)
+
+
+def test_eligible_author_pooled_with_elderly_fields():
+    items = [_item("A", "c_a")]
+    decisions = [_decision("c_a", "ok", band_percent=36.86, band_tgi=155.72)]
+    _summaries, asset_rows, role_rows = _run(items, decisions)
+    assert len(asset_rows) == 1
+    row = asset_rows[0]
+    assert row["platform_author_id"] == "A"
+    assert row["elderly_ratio"] == 36.86
+    assert row["elderly_tgi"] == 155.72
+    assert row["eligible_as_source"] == 1
+    assert "high_50plus_profile" in _summaries[0]["roles"]
+    assert any(r["role"] == "high_50plus_profile" for r in role_rows)
+
+
+def test_ratio_below_30_rejected():
+    items = [_item("B", "c_b")]
+    decisions = [_decision("c_b", "ok", band_percent=22.21, band_tgi=140)]  # 占比<30
+    _s, asset_rows, _r = _run(items, decisions)
+    assert asset_rows == []
+
+
+def test_tgi_below_120_rejected():
+    items = [_item("C", "c_c")]
+    decisions = [_decision("c_c", "ok", band_percent=40, band_tgi=100)]  # TGI<120
+    _s, asset_rows, _r = _run(items, decisions)
+    assert asset_rows == []
+
+
+def test_non_douyin_no_portrait_rejected():
+    items = [_item("D", "c_d", platform="kuaishou")]
+    decisions = [_decision("c_d", fifty_status=None)]  # 无 fifty_plus → 不收
+    _s, asset_rows, _r = _run(items, decisions)
+    assert asset_rows == []
+
+
+def test_boundary_exactly_30_and_120_excluded():
+    # 严格大于:占比=30、TGI=120 都不算达标。
+    items = [_item("E", "c_e")]
+    decisions = [_decision("c_e", "ok", band_percent=30, band_tgi=120)]
+    _s, asset_rows, _r = _run(items, decisions)
+    assert asset_rows == []

tests/test_douyin_fifty_plus_scoring.py

@@ -55,7 +55,7 @@ def _decide(bundle):
 
 
 def test_douyin_ok_uses_35_35_30():
-    fifty = {"status": "ok", "score": 76, "components": {"band": "41+", "band_tgi": 171.5}}
+    fifty = {"status": "ok", "score": 76, "components": {"band": "50+", "band_tgi": 171.5}}
     decision = _decide(_bundle(80, 70, fifty))
     scorecard = decision["scorecard"]
     assert scorecard["fifty_plus_status"] == "ok"

tests/test_fifty_plus_score_formula.py

@@ -1,8 +1,8 @@
-"""M9E:50+ 子分公式——band 41+、TGI 200 满分、TGI 6 成 + 占比 4 成、缺字段 incomplete。"""
+"""50+ 子分公式——band 纯50+(仅 50- 桶,41-50 已忽略)、TGI 200 满分、TGI 6 成 + 占比 4 成、缺字段 incomplete。"""
 
 from content_agent.business_modules.content_discovery.fifty_plus import fifty_plus_score
 
-_OLD = {  # 广场舞健身号(强 41+)
+_OLD = {  # 广场舞健身号:41-50 应被忽略,只算 50- 桶
     "41-50": {"percentage": "31.73%", "preference": "210.02"},
     "50-": {"percentage": "29.76%", "preference": "130.53"},
 }
@@ -16,9 +16,11 @@ def test_score_separates_old_from_young():
     old = fifty_plus_score(_OLD)
     young = fifty_plus_score(_YOUNG)
     assert old["status"] == "ok" and young["status"] == "ok"
-    assert old["band"] == "41+"
-    assert 75 <= old["score"] <= 77          # ≈76
-    assert 10 <= young["score"] <= 12         # ≈11
+    assert old["band"] == "50+"
+    # 纯 50+ 口径:只算 50- 桶(29.76% / TGI 130.53)→ 0.6*65.27+0.4*29.76 ≈ 51.06
+    assert old["band_percent"] == 29.76
+    assert 50 <= old["score"] <= 52
+    assert 6 <= young["score"] <= 8           # 仅 50-=5.5%/17 → ≈7.3
     assert old["score"] > young["score"]
 
 

tests/test_p7_author_assets.py

@@ -22,7 +22,7 @@ class CapturingRuntimeStore(LocalRuntimeFileStore):
         self.publish_jobs.extend(rows)
 
 
-def test_author_assets_are_written_only_when_sampling_rules_pass(tmp_path):
+def test_author_assets_are_written_only_when_elderly_portrait_rules_pass(tmp_path):
     runtime = CapturingRuntimeStore(tmp_path / "runtime")
     runtime.prepare_run("run_001")
 
@@ -113,7 +113,10 @@ def _decisions():
                 "strategy_version": "V1",
                 "search_query_effect_status": "success" if index <= 3 else "pending",
                 "source_evidence": {"source_kind": "pattern_itemset"},
-                "age_50_plus_level": "medium",
+                "scorecard": {
+                    "fifty_plus_status": "ok",
+                    "fifty_plus_components": {"band": "50+", "band_percent": 36.0, "band_tgi": 150.0},
+                },
             }
         )
     return decisions