124 lines
4.6 KiB
Python
124 lines
4.6 KiB
Python
import argparse
|
|
import json
|
|
|
|
|
|
def current_set() -> int:
|
|
from tft import paths
|
|
|
|
pointer = paths.latest_static_pointer()
|
|
if not pointer.exists():
|
|
raise SystemExit("no static data yet: run `fetch-static` first")
|
|
return json.loads(pointer.read_text())["set"]
|
|
|
|
|
|
def main() -> None:
|
|
parser = argparse.ArgumentParser(prog="tft")
|
|
sub = parser.add_subparsers(dest="command", required=True)
|
|
|
|
p_fetch = sub.add_parser("fetch-static", help="download static data from Community Dragon")
|
|
p_fetch.add_argument("--set", type=int, default=None, help="override set number")
|
|
|
|
sub.add_parser("crawl", help="crawl ranked matches from Challenger/GM ladder (max 200 neue pro Lauf)")
|
|
|
|
sub.add_parser("extract", help="extract endboards from raw matches")
|
|
sub.add_parser("build-artifact", help="build analysis.json from static data + endboards")
|
|
sub.add_parser("calibrate", help="backtest score vs real placements (holdout)")
|
|
|
|
p_auto = sub.add_parser("autoplay", help="run scripted games headless")
|
|
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":
|
|
from tft.staticdata.fetch import fetch_static
|
|
|
|
parsed = fetch_static(set_override=args.set)
|
|
meta = parsed["meta"]
|
|
print(f"set {meta['set']} ({meta['set_name']}), patch {meta['patch']}")
|
|
for name in ("units", "traits", "items", "augments"):
|
|
print(f" {name}: {len(parsed[name])}")
|
|
|
|
elif args.command == "crawl":
|
|
from tft.matches.crawl import crawl
|
|
|
|
added = crawl(set_number=current_set())
|
|
print(f"{added} new matches stored")
|
|
|
|
elif args.command == "extract":
|
|
from tft import db
|
|
from tft.matches.extract import extract_all
|
|
from tft.staticdata.fetch import load_static
|
|
|
|
conn = db.connect()
|
|
n, unknown = extract_all(conn, load_static())
|
|
conn.close()
|
|
print(f"{n} endboards extracted")
|
|
if unknown:
|
|
print("UNKNOWN IDS (patch drift?):")
|
|
for key, count in unknown.most_common():
|
|
print(f" {key}: {count}")
|
|
else:
|
|
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()
|
|
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} ({n_boards} boards learned from)")
|
|
|
|
elif args.command == "calibrate":
|
|
from tft import db
|
|
from tft.model import artifact as artifact_mod
|
|
from tft.model.calibrate import calibrate
|
|
from tft.staticdata.fetch import load_static
|
|
|
|
conn = db.connect()
|
|
learned_art = artifact_mod.load(current_set())
|
|
baseline_art = artifact_mod.build(load_static(), {})
|
|
r_base = calibrate(conn, baseline_art, holdout_only=True)
|
|
r_learned = calibrate(conn, learned_art, holdout_only=True)
|
|
conn.close()
|
|
print(f"holdout matches: {r_learned['matches']}")
|
|
print(f"spearman baseline: {r_base['mean_spearman']:.3f}")
|
|
print(f"spearman learned: {r_learned['mean_spearman']:.3f}")
|
|
|
|
elif args.command == "autoplay":
|
|
from tft.constants.loader import load_constants
|
|
from tft.model import artifact as artifact_mod
|
|
from tft.sim.autoplay import run
|
|
|
|
set_number = current_set()
|
|
result = run(
|
|
artifact_mod.load(set_number), load_constants(set_number),
|
|
args.policy, args.games,
|
|
)
|
|
print(f"{args.policy}: {result['games']} games, "
|
|
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()
|