M6+M11: learned weights with shrinkage, refresh command, cron docs

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

backend/tests/test_learn.py

@@ -0,0 +1,52 @@
+import json
+
+from tft.model.learn import learn
+
+
+def board_row(placement, unit="TFT17_A", item="TFT_Item_X", trait_tier=1):
+    units = [{"character_id": unit, "tier": 2, "itemNames": [item]}]
+    traits = [{"name": "TFT17_T", "num_units": 2, "tier_current": trait_tier}]
+    return (placement, json.dumps(units), json.dumps(traits), json.dumps([]))
+
+
+def test_strong_item_gets_multiplier_above_1():
+    rows = [board_row(1, item="TFT_Item_Good") for _ in range(200)]
+    rows += [board_row(8, item="TFT_Item_Bad") for _ in range(200)]
+    learned = learn(rows)
+    assert learned["items"]["TFT_Item_Good"] > 1.05
+    assert learned["items"]["TFT_Item_Bad"] < 0.95
+
+
+def test_shrinkage_with_thin_data():
+    rows = [board_row(1, item="TFT_Item_Rare") for _ in range(3)]
+    learned = learn(rows)
+    assert abs(learned["items"]["TFT_Item_Rare"] - 1.0) < 0.06
+
+
+def test_trait_keys_match_score_format():
+    rows = [board_row(2, trait_tier=3) for _ in range(50)]
+    learned = learn(rows)
+    assert "TFT17_T@3" in learned["traits"]
+
+
+def test_pair_lift_for_cooccurring_units():
+    def pair_row(placement):
+        units = [
+            {"character_id": "TFT17_A", "tier": 2, "itemNames": []},
+            {"character_id": "TFT17_B", "tier": 2, "itemNames": []},
+        ]
+        return (placement, json.dumps(units), json.dumps([]), json.dumps([]))
+
+    # A und B stehen immer zusammen in Top-4-Boards; C ist überall.
+    rows = [pair_row(1) for _ in range(100)]
+    solo = [{"character_id": "TFT17_C", "tier": 1, "itemNames": []}]
+    rows += [(1, json.dumps(solo), json.dumps([]), json.dumps([])) for _ in range(100)]
+    learned = learn(rows)
+    assert learned["pairs"]["TFT17_A|TFT17_B"] > 0.5
+
+
+def test_empty_rows():
+    learned = learn([])
+    assert learned["traits"] == {}
+    assert learned["pairs"] == {}
+    assert learned["n_boards"] == 0

backend/tft/cli.py

@@ -29,6 +29,8 @@ def main() -> None:
     p_auto.add_argument("--policy", choices=["afk", "econ"], default="econ")
     p_auto.add_argument("--games", type=int, default=200)
 
+    sub.add_parser("refresh", help="daily job: crawl + extract + build-artifact + calibrate")
+
     args = parser.parse_args()
 
     if args.command == "fetch-static":
@@ -63,14 +65,19 @@ def main() -> None:
             print("all ids resolved against static data")
 
     elif args.command == "build-artifact":
+        from tft import db
         from tft.model import artifact as artifact_mod
+        from tft.model.learn import learn_from_db
         from tft.staticdata.fetch import load_static
 
         static = load_static()
-        learned = {}
-        art = artifact_mod.build(static, learned)
+        conn = db.connect()
+        learned = learn_from_db(conn, static["meta"]["set"])
+        conn.close()
+        n_boards = learned.pop("n_boards", 0)
+        art = artifact_mod.build(static, learned, {"boards_learned_from": n_boards})
         path = artifact_mod.save(art)
-        print(f"artifact written: {path}")
+        print(f"artifact written: {path} ({n_boards} boards learned from)")
 
     elif args.command == "calibrate":
         from tft import db
@@ -98,6 +105,16 @@ def main() -> None:
               f"avg placement {result['avg_placement']:.2f}, "
               f"top4 {result['top4_rate']:.0%}")
 
+    elif args.command == "refresh":
+        import subprocess
+        import sys
+
+        for stage in ("crawl", "extract", "build-artifact", "calibrate"):
+            print(f"== {stage} ==", flush=True)
+            result = subprocess.run([sys.executable, "-m", "tft.cli", stage])
+            if result.returncode != 0:
+                raise SystemExit(f"refresh aborted at {stage}")
+
 
 if __name__ == "__main__":
     main()

backend/tft/model/learn.py

@@ -0,0 +1,88 @@
+"""Learn multiplicative strength weights from endboards via placement deltas.
+
+All multipliers center at 1.0 with Bayesian shrinkage: with thin data the
+score model falls back to the rule baseline (survives set launch).
+"""
+
+import json
+from collections import defaultdict
+
+MEAN_PLACEMENT = 4.5
+PSEUDO_COUNT = 30
+# Platzierungs-Delta -> Multiplikator: 1 Platz besser als der Schnitt ≈ +6 %.
+DELTA_SCALE = 0.06
+MULT_CAP = (0.8, 1.25)
+PAIR_MIN_N = 20
+PAIR_LIFT_CAP = (-0.9, 1.15)
+
+
+def _multiplier(placements: list[int]) -> float:
+    n = len(placements)
+    mean = (sum(placements) + MEAN_PLACEMENT * PSEUDO_COUNT) / (n + PSEUDO_COUNT)
+    mult = 1.0 + (MEAN_PLACEMENT - mean) * DELTA_SCALE
+    return min(max(mult, MULT_CAP[0]), MULT_CAP[1])
+
+
+def learn(rows: list[tuple]) -> dict:
+    """rows: (placement, units_json, traits_json, augments_json)."""
+    trait_placements = defaultdict(list)
+    item_placements = defaultdict(list)
+    augment_placements = defaultdict(list)
+    unit_top4 = defaultdict(int)
+    pair_top4 = defaultdict(int)
+    n_boards = 0
+    n_top4 = 0
+
+    for placement, units_json, traits_json, augments_json in rows:
+        n_boards += 1
+        top4 = placement <= 4
+        n_top4 += top4
+
+        units = json.loads(units_json)
+        names = sorted({u["character_id"] for u in units})
+        for u in units:
+            for item in u.get("itemNames", []):
+                item_placements[item].append(placement)
+        for t in json.loads(traits_json):
+            if t.get("tier_current", 0) > 0:
+                trait_placements[f"{t['name']}@{t['tier_current']}"].append(placement)
+        for a in json.loads(augments_json):
+            augment_placements[a].append(placement)
+
+        if top4:
+            for name in names:
+                unit_top4[name] += 1
+            for i, a in enumerate(names):
+                for b in names[i + 1 :]:
+                    pair_top4[(a, b)] += 1
+
+    pairs = {}
+    if n_top4:
+        for (a, b), n_ab in pair_top4.items():
+            if n_ab < PAIR_MIN_N:
+                continue
+            expected = unit_top4[a] * unit_top4[b] / n_top4
+            if expected <= 0:
+                continue
+            lift = n_ab / expected - 1.0
+            lift = min(max(lift, PAIR_LIFT_CAP[0]), PAIR_LIFT_CAP[1])
+            pairs[f"{a}|{b}"] = round(lift, 4)
+
+    return {
+        "traits": {k: round(_multiplier(v), 4) for k, v in trait_placements.items()},
+        "items": {k: round(_multiplier(v), 4) for k, v in item_placements.items()},
+        "augments": {k: round(_multiplier(v), 4) for k, v in augment_placements.items()},
+        "pairs": pairs,
+        "n_boards": n_boards,
+    }
+
+
+def learn_from_db(conn, set_number: int, exclude_holdout: bool = True) -> dict:
+    where = "WHERE set_number = ?"
+    if exclude_holdout:
+        where += " AND rowid % 5 != 0"
+    rows = conn.execute(
+        f"SELECT placement, units, traits, augments FROM endboards {where}",
+        (set_number,),
+    ).fetchall()
+    return learn(rows)

backend/tft/model/score.py

@@ -62,8 +62,11 @@ def score_board(board_units: list[dict], augments: list[str], artifact: dict) ->
         total *= augment_mults.get(augment, 1.0)
 
     names = sorted({u["api_name"] for u in board_units})
-    for i, a in enumerate(names):
-        for b in names[i + 1 :]:
-            total += pair_lifts.get(f"{a}|{b}", 0.0)
+    lift_sum = sum(
+        pair_lifts.get(f"{a}|{b}", 0.0)
+        for i, a in enumerate(names)
+        for b in names[i + 1 :]
+    )
+    total *= 1 + min(max(lift_sum * 0.01, -0.10), 0.10)
 
     return total