"""MASLDTriangulate-Kor — CLI entrypoint (stdlib only). Usage: python3 main.py --help python3 main.py --grid python3 main.py --pair "MASH" "CV death" python3 main.py --mediation "MASH" "CV death" python3 main.py --genotype "PNPLA3" "resmetirom" "MASH" python3 main.py --discordant --top 5 python3 main.py --korean python3 main.py --report out.json python3 main.py --report-md out.md python3 main.py --designs python3 main.py --taxonomy """ from __future__ import annotations import argparse import json import sys from typing import Optional from triangulation import DISCLAIMER from triangulation.bias import diagnose_discordance, taxonomy from triangulation.designs import proposed_designs from triangulation.genotype import subgroup_treatment_effect from triangulation.grid import build_grid, discordant_top, grid_as_table, design_summary from triangulation.lawlor import score_pair, korean_ancestry_layer_score from triangulation.mediation import masld_mediated_fraction from triangulation.mvmr import mvmr_decompose from triangulation.ontology import load_effects, load_outcomes, load_stages, load_instruments def _print_disclaimer(): print(DISCLAIMER) print() def cmd_grid(args, effects): grid = build_grid(effects) _print_disclaimer() print(f"=== 5-design grid ({len(grid)} pairs) ===") rows = grid_as_table(grid) widths = [max(len(str(r[i])) for r in rows) for i in range(len(rows[0]))] for i, r in enumerate(rows): line = " ".join(str(c).ljust(widths[idx]) for idx, c in enumerate(r)) print(line) if i == 0: print("-" * (sum(widths) + 2 * (len(widths) - 1))) def cmd_pair(args, effects): stage, outcome = args.pair summary = design_summary(effects, stage, outcome) from triangulation.grid import concordance_score conc = concordance_score(summary) law = score_pair(summary) _print_disclaimer() print(f"=== Pair: {stage} × {outcome} ===") print(f"Concordance: {conc['label']} (score {conc['score']}, n_designs {conc['n_designs']})") print(f"Lawlor 5-criterion: {law['total']}/5 {law['criteria']}") print() for d, v in summary.items(): print(f"[{d}] n={v['n']} effect={v['effect']} CI=({v['ci_low']}, {v['ci_high']}) " f"dir={v['direction']} sig={v['significant']}") def cmd_mediation(args, effects): stage, outcome = args.mediation med = masld_mediated_fraction(effects, stage, outcome) mv = mvmr_decompose(effects, stage, outcome) _print_disclaimer() print(f"=== Mediation: {stage} × {outcome} ===") print(json.dumps(med, ensure_ascii=False, indent=2, default=str)) print() print("[MVMR decomposition]") print(json.dumps(mv, ensure_ascii=False, indent=2, default=str)) def cmd_genotype(args, effects): gene, drug, outcome = args.genotype res = subgroup_treatment_effect(drug, gene, outcome) _print_disclaimer() print(f"=== Genotype subgroup: {drug} × {gene} → {outcome} ===") print(json.dumps(res, ensure_ascii=False, indent=2, default=str)) def cmd_discordant(args, effects): grid = build_grid(effects) top = discordant_top(grid, top_n=int(args.top)) _print_disclaimer() print(f"=== Top {len(top)} discordant pairs ===") for g in top: print() print(f"[{g['stage']} × {g['outcome']}] {g['concordance']}") diag = diagnose_discordance(effects, g["stage"], g["outcome"]) for f in diag["flags"]: print(f" - {f['code']}: {f['trigger']}") def cmd_korean(args, effects): layer = korean_ancestry_layer_score(effects) _print_disclaimer() print("=== Korean ancestry / lean MASLD layer ===") print(json.dumps(layer, ensure_ascii=False, indent=2, default=str)) print() # also break down korean rows korean = [e for e in effects if (e.get("ancestry") or "").lower() == "korean"] print(f"Korean rows ({len(korean)}):") for r in korean: print(f" {r.get('masld_stage')} × {r.get('outcome')} [{r.get('design')}] " f"effect={r.get('effect_estimate')} ({r.get('source_citation')})") def cmd_report(args, effects): from triangulation.report import build_full_report, report_to_json rep = build_full_report(effects) txt = report_to_json(rep) with open(args.report, "w", encoding="utf-8") as f: f.write(txt) print(f"saved JSON report → {args.report} ({len(txt)} bytes)") def cmd_report_md(args, effects): from triangulation.report import build_full_report, report_to_markdown rep = build_full_report(effects) md = report_to_markdown(rep) with open(args.report_md, "w", encoding="utf-8") as f: f.write(md) print(f"saved Markdown report → {args.report_md} ({len(md)} bytes)") def cmd_designs(args, effects): res = proposed_designs(effects) _print_disclaimer() print(f"=== 8 triangulation-targeted design cards ===") for c in res["cards"]: print() print(f"[{c['tier']}] {c['type']}") print(f" rationale: {c['rationale']}") print(f" endpoint: {c['primary_endpoint']}") print(f" n: {c['sample_size']} follow-up {c['follow_up']}") print(f" hypothesis:{c['key_hypothesis']}") if c.get("mr_instruments_required"): print(f" IVs: {c['mr_instruments_required']}") if c.get("genotype_stratification"): print(" *** genotype-stratified") def cmd_taxonomy(args, effects): _print_disclaimer() print("=== Bias taxonomy ===") for b in taxonomy(): print() print(f"[{b['code']}] {b['name']}") print(f" {b['description']}") print(f" remedy: {b['remedy']}") def cmd_data_test(args, effects): out = { "effects_loaded": len(effects), "stages_loaded": len(load_stages()), "outcomes_loaded": len(load_outcomes()), "instruments_loaded": len(load_instruments()), "designs_unique": sorted({e.get("design") for e in effects if e.get("design")}), "stages_unique": sorted({e.get("masld_stage") for e in effects if e.get("masld_stage")}), "outcomes_unique_count": len({e.get("outcome") for e in effects if e.get("outcome")}), "korean_rows": sum(1 for e in effects if (e.get("ancestry") or "").lower() == "korean"), } print(json.dumps(out, ensure_ascii=False, indent=2, default=str)) def build_parser(): p = argparse.ArgumentParser( prog="masld-triangulate-kor", description="MASLDTriangulate-Kor: MASLD 단계 × multi-outcome 5-design 인과추론 triangulation", ) p.add_argument("--effects", default=None, help="custom effects CSV path") g = p.add_mutually_exclusive_group() g.add_argument("--grid", action="store_true", help="5-design grid 전체 출력") g.add_argument("--pair", nargs=2, metavar=("STAGE", "OUTCOME"), help="단일 pair 상세") g.add_argument("--mediation", nargs=2, metavar=("STAGE", "OUTCOME"), help="MASLD-mediated vs metabolic-confounder-mediated 분리") g.add_argument("--genotype", nargs=3, metavar=("GENE", "DRUG", "OUTCOME"), help="genotype-stratified treatment effect 가설") g.add_argument("--discordant", action="store_true", help="discordant pair 상위 분석") g.add_argument("--korean", action="store_true", help="한국 ancestry / lean MASLD layer") g.add_argument("--report", metavar="OUT.json", help="full JSON 리포트 저장") g.add_argument("--report-md", dest="report_md", metavar="OUT.md", help="Markdown 리포트 저장") g.add_argument("--designs", action="store_true", help="8개 후속 design 카드") g.add_argument("--taxonomy", action="store_true", help="bias taxonomy 출력") g.add_argument("--data-test", action="store_true", help="데이터 로드 sanity check") p.add_argument("--top", default=5, type=int, help="--discordant 시 표시할 pair 개수") return p def main(argv=None): parser = build_parser() args = parser.parse_args(argv) effects = load_effects(args.effects) if not effects: print("ERROR: effects CSV 로드 실패.", file=sys.stderr) return 2 if args.grid: cmd_grid(args, effects) elif args.pair: cmd_pair(args, effects) elif args.mediation: cmd_mediation(args, effects) elif args.genotype: cmd_genotype(args, effects) elif args.discordant: cmd_discordant(args, effects) elif args.korean: cmd_korean(args, effects) elif args.report: cmd_report(args, effects) elif args.report_md: cmd_report_md(args, effects) elif args.designs: cmd_designs(args, effects) elif args.taxonomy: cmd_taxonomy(args, effects) elif args.data_test: cmd_data_test(args, effects) else: parser.print_help() return 0 if __name__ == "__main__": sys.exit(main())