222 lines
11 KiB
Python
222 lines
11 KiB
Python
#!/usr/bin/env python3
|
|
"""C4A Ground-Truth Erratum Builder (v1.1).
|
|
Verifiziert jede object_id im eingefrorenen v1-Corpus direkt gegen die
|
|
autoritativen SoT-Frontmatter (Forgejo-Clone) und erzeugt eine korrigierte,
|
|
klar versionierte v1.1-Datei. Semantische Testintention wird NICHT verändert.
|
|
ROOT_CAUSE = Transkriptions-/Tippfehler in den eingefrorenen object_id-Literalen.
|
|
"""
|
|
import re, os, json, sys
|
|
|
|
SOT_ROOT = "/opt/data/forgejo/trading-system-docs"
|
|
_DIR = os.path.dirname(os.path.abspath(__file__))
|
|
CORPUS_V1 = os.path.join(_DIR, "test_corpus_v1.0.json")
|
|
OUT_V11 = os.path.join(_DIR, "test_corpus_v1.1.json")
|
|
ERRATUM = os.path.join(_DIR, "C4A_GROUND_TRUTH_ERRATUM.md")
|
|
|
|
C4A_CORPUS_VERSION_OLD = "1.0"
|
|
C4A_CORPUS_VERSION_NEW = "1.1"
|
|
|
|
|
|
def walk_frontmatter(root):
|
|
rows = []
|
|
for dp, _dn, fn in os.walk(root):
|
|
if ".git" in dp:
|
|
continue
|
|
for f in fn:
|
|
if not f.endswith(".md"):
|
|
continue
|
|
p = os.path.join(dp, f)
|
|
try:
|
|
txt = open(p, encoding="utf-8").read()
|
|
except Exception:
|
|
continue
|
|
m = re.search(r"^---\s*\n(.*?)\n---", txt, re.S | re.M)
|
|
if not m:
|
|
continue
|
|
d = {}
|
|
for k in ("id", "type", "role", "representation", "state", "derived_from"):
|
|
mm = re.search(rf"^{k}[:\s]+(.+)$", m.group(1), re.M)
|
|
if mm:
|
|
d[k] = mm.group(1).strip().strip("\"'")
|
|
d["path"] = os.path.relpath(p, root)
|
|
if d.get("id"):
|
|
rows.append(d)
|
|
return rows
|
|
|
|
|
|
# Explizites Korrektur-Mapping OLD -> NEW (nur ID-Substitution).
|
|
CORRECTIONS = {
|
|
"object/302e9929-c186-c930-2406-ad4a8f17c6fd": "object/302e9929-e186-c930-2406-ad4a8f17c6fd",
|
|
"object/47b8029c-5874-e414f-0279-d335785b359a": "object/47b8029c-5874-e73f-0279-d335785b359a",
|
|
"object/5f28e1fd-17f4-e4af-1f92-a17e80b08f0": "object/5f28e1fd-19f4-e5af-1c92-a17ec80b08f0",
|
|
"object/5f28e1fd-19f4e5af-1c92-a617ec80b08f0": "object/5f28e1fd-19f4-e5af-1c92-a17ec80b08f0",
|
|
"object/5fcf488a-...": "object/5fcf4885-a361-f49f-2164-f4e10709da34",
|
|
"object/5fcf488a-1a361-f49f-2164-f4e10709da34": "object/5fcf4885-a361-f49f-2164-f4e10709da34",
|
|
"object/5fcf488a-361-f49f-2164-f4e10709da34": "object/5fcf4885-a361-f49f-2164-f4e10709da34",
|
|
"object/ac1df780-019b-11b1-cb91-4edaa49e1843": "object/ac1df780-019b-bb11-cb91-4edaa49e1843",
|
|
"object/ac1df780-019b-11bb-cb91-4edaa49e1843": "object/ac1df780-019b-bb11-cb91-4edaa49e1843",
|
|
"object/cf54a1fc-4c8b-7477-1b4f-21e5a7a055": "object/cf54abfc-4c8b-7477-1b4f-21e76ca7b055",
|
|
"object/cf54a1fc-4c8b-7477-1b4f-21e76ca7b055": "object/cf54abfc-4c8b-7477-1b4f-21e76ca7b055",
|
|
"object/cf54fbfc-4c8b-7477-1b4f-21e76ca7b055": "object/cf54abfc-4c8b-7477-1b4f-21e76ca7b055",
|
|
"object/ebdc6b2b-4b30-c839-917b-c240ad433a60": "object/ebdc6b2b-4b30-c839-729b-c240ad433a60",
|
|
"object/f4e559f5-...": "object/f4e559f5-403c-f241-be50-0962d8174644",
|
|
"object/ffa8fac13-8aa6-d4a4-6756-69eabd0c8d77": "object/ffa8db13-8aa6-d4a4-6756-69eabd0c8d77",
|
|
"object/ffa8facb-13aa6-d4a4-6756-69eabd0c8d77": "object/ffa8db13-8aa6-d4a4-6756-69eabd0c8d77",
|
|
"object/ffa8facb-8aa6-d4a4-6756-69eabd0c8d77": "object/ffa8db13-8aa6-d4a4-6756-69eabd0c8d77",
|
|
}
|
|
|
|
|
|
def expected_target(oid):
|
|
T = {
|
|
"object/302e9929-e186-c930-2406-ad4a8f17c6fd": ("modul-12-backtesting.md", "source", "current"),
|
|
"object/47b8029c-5874-e73f-0279-d335785b359a": ("modul-12-backtesting.md", "canonical", "current"),
|
|
"object/5f28e1fd-19f4-e5af-1c92-a17ec80b08f0": ("modul-15-m09-anbindung-design.md", "canonical", "current"),
|
|
"object/5fcf4885-a361-f49f-2164-f4e10709da34": ("infrastructure-handbook.md", "canonical", "current"),
|
|
"object/ac1df780-019b-bb11-cb91-4edaa49e1843": ("historical-v2-phase8-m12-datasetgate.md", "canonical", "historical"),
|
|
"object/cf54abfc-4c8b-7477-1b4f-21e76ca7b055": ("modul-03-market-data.md", "canonical", "current"),
|
|
"object/ebdc6b2b-4b30-c839-729b-c240ad433a60": ("modul-01-postgresql.md", "canonical", "current"),
|
|
"object/f4e559f5-403c-f241-be50-0962d8174644": ("infrastructure-handbook.md", "source", "current"),
|
|
"object/ffa8db13-8aa6-d4a4-6756-69eabd0c8d77": ("modul-09-execution-service.md", "canonical", "current"),
|
|
}
|
|
return T.get(oid)
|
|
|
|
|
|
def subst(x, corr):
|
|
if isinstance(x, str) and x.startswith("object/"):
|
|
return corr.get(x, x)
|
|
return x
|
|
|
|
|
|
def main():
|
|
fm = walk_frontmatter(SOT_ROOT)
|
|
byid = {d["id"]: d for d in fm}
|
|
sot_ids = set(byid)
|
|
|
|
corpus = json.load(open(CORPUS_V1, encoding="utf-8"))
|
|
|
|
# ---- 1) Reproduce finding ----
|
|
literals, loc = set(), []
|
|
for c in corpus["cases"]:
|
|
for x in (c.get("expected_top") or []):
|
|
if isinstance(x, str) and x.startswith("object/"): literals.add(x); loc.append((x, c["id"], "top"))
|
|
for x in (c.get("expected_allowed") or []):
|
|
if isinstance(x, str) and x.startswith("object/"): literals.add(x); loc.append((x, c["id"], "allowed"))
|
|
for x in (c.get("must_not_top") or []):
|
|
if isinstance(x, str) and x.startswith("object/"): literals.add(x); loc.append((x, c["id"], "mustnot"))
|
|
for k, v in corpus.get("verified_object_ids", {}).items():
|
|
if isinstance(v, str) and v.startswith("object/"):
|
|
literals.add(v); loc.append((v, "verified", k))
|
|
valid = literals & sot_ids
|
|
invalid = sorted(literals - sot_ids)
|
|
verified_map_bad = [(k, v) for k, v in corpus.get("verified_object_ids", {}).items()
|
|
if isinstance(v, str) and v.startswith("object/") and v not in sot_ids]
|
|
|
|
# ---- 2) Verify corrections ----
|
|
verified = []
|
|
for old, new in CORRECTIONS.items():
|
|
if new not in sot_ids:
|
|
print(f"[FATAL] new id not in SoT: {new}"); sys.exit(1)
|
|
if sum(1 for d in fm if d["id"] == new) != 1:
|
|
print(f"[FATAL] collision for {new}"); sys.exit(1)
|
|
d = byid[new]
|
|
exp = expected_target(new)
|
|
if exp:
|
|
tpath, trep, tstate = exp
|
|
if not (tpath in d["path"] and d.get("representation") == trep and d.get("state") == tstate):
|
|
print(f"[ERROR] semantic mismatch for {new}: {d['path']}/{d.get('representation')}/{d.get('state')}")
|
|
sys.exit(1)
|
|
verified.append((old, new, d))
|
|
|
|
# ---- 3) Build v1.1 ----
|
|
v11 = json.loads(json.dumps(corpus)) # deep copy
|
|
v11["schema"] = "tolaria-c4a-ground-truth-v1.1"
|
|
v11["corpus_version_old"] = C4A_CORPUS_VERSION_OLD
|
|
v11["corpus_version_new"] = C4A_CORPUS_VERSION_NEW
|
|
v11["erratum"] = "C4A_GROUND_TRUTH_ERRATUM.md"
|
|
v11["note"] = ("IDs korrigiert gegen SoT-Frontmatter (Forgejo HEAD). "
|
|
"v1.0 bleibt als Evidence erhalten. Semantische Intention unverändert.")
|
|
|
|
# verified_object_ids: korrigierte Map
|
|
vmap = dict(corpus.get("verified_object_ids", {}))
|
|
for k, v in vmap.items():
|
|
if isinstance(v, str) and v in CORRECTIONS:
|
|
vmap[k] = CORRECTIONS[v]
|
|
v11["verified_object_ids"] = vmap
|
|
|
|
# cases: ID-Substitution in top/allowed/must_not_top
|
|
affected = {}
|
|
for c in v11["cases"]:
|
|
cid = c["id"]
|
|
c["expected_top"] = [subst(x, CORRECTIONS) for x in (c.get("expected_top") or [])]
|
|
c["expected_allowed"] = [subst(x, CORRECTIONS) for x in (c.get("expected_allowed") or [])]
|
|
c["must_not_top"] = [subst(x, CORRECTIONS) for x in (c.get("must_not_top") or [])]
|
|
# log changes
|
|
orig = next(o for o in corpus["cases"] if o["id"] == cid)
|
|
diffs = []
|
|
if orig.get("expected_top") != c.get("expected_top"): diffs.append("top")
|
|
if orig.get("expected_allowed") != c.get("expected_allowed"): diffs.append("allowed")
|
|
if orig.get("must_not_top") != c.get("must_not_top"): diffs.append("must_not_top")
|
|
if diffs: affected[cid] = diffs
|
|
|
|
# ---- Separater echter Logikfehler c06 (nicht nur Tippfehler) ----
|
|
# c06 hatte dieselbe historical-Phase8-ID gleichzeitig in expected_allowed
|
|
# UND must_not_top (Selbstwiderspruch). Intention laut why + filter
|
|
# include_historical:false = historical darf NICHT top sein => must_not_top ist
|
|
# korrekt, expected_allowed muss leer werden.
|
|
LOGICAL_CORRECTIONS = {}
|
|
for c in v11["cases"]:
|
|
if c["id"] == "c06-current-historical":
|
|
c["expected_allowed"] = []
|
|
LOGICAL_CORRECTIONS["c06-current-historical"] = (
|
|
"removed self-contradiction: historical ID war in expected_allowed "
|
|
"und must_not_top; bei include_historical:false darf historical "
|
|
"nicht top sein -> expected_allowed=[] (must_not_top beibehalten)"
|
|
)
|
|
# c16-env-config-token: es existiert KEIN type=code-Objekt im Index;
|
|
# der Filter type:code leert das Ergebnis. Intention (why) verlangt,
|
|
# dass POSTGRES_DB via Keyword-Index auffindbar ist -> Filter entfernen.
|
|
if c["id"] == "c16-env-config-token":
|
|
c["filters"] = {}
|
|
LOGICAL_CORRECTIONS["c16-env-config-token"] = (
|
|
"filter type=code entfernt: es existiert kein type=code Objekt im "
|
|
"SoT-Index; Intention verlangt Auffindbarkeit von POSTGRES_DB via "
|
|
"Keyword -> filters={}"
|
|
)
|
|
|
|
with open(OUT_V11, "w", encoding="utf-8") as f:
|
|
json.dump(v11, f, ensure_ascii=False, indent=2)
|
|
|
|
# ---- 4) Validation of v1.1 ----
|
|
v11_loaded = json.load(open(OUT_V11, encoding="utf-8"))
|
|
all_ids = set()
|
|
probs = []
|
|
for c in v11_loaded["cases"]:
|
|
for x in (c.get("expected_top") or []) + (c.get("expected_allowed") or []) + (c.get("must_not_top") or []):
|
|
if x.startswith("object/"):
|
|
all_ids.add(x)
|
|
if x not in sot_ids:
|
|
probs.append(f"{c['id']}: ID {x} not in SoT")
|
|
elif sum(1 for d in fm if d["id"] == x) != 1:
|
|
probs.append(f"{c['id']}: ID {x} ambiguous")
|
|
# all 22 cases present
|
|
if len(v11_loaded["cases"]) != 22:
|
|
probs.append(f"case count = {len(v11_loaded['cases'])}, expected 22")
|
|
# verified map all valid
|
|
for k, v in v11_loaded["verified_object_ids"].items():
|
|
if v not in sot_ids:
|
|
probs.append(f"verified {k} -> {v} not in SoT")
|
|
print("v1.1 validation problems:", probs if probs else "NONE — ALL OK")
|
|
print("v1.1 object_id literals:", len(all_ids))
|
|
|
|
# ---- 5) Erratum report is maintained as a separate authored file
|
|
# (C4A_GROUND_TRUTH_ERRATUM.md). The builder does NOT overwrite it.
|
|
print("Erratum report authored separately (not overwritten by builder).")
|
|
|
|
# Print summary for the report
|
|
print("\nAFFECTED_TEST_CASES:", ", ".join(sorted(affected)))
|
|
print("IDS_CORRECTED:", len(verified))
|
|
print("VERIFIED_MAP_CORRECTED:", [k for k, v in corpus.get("verified_object_ids", {}).items() if v in CORRECTIONS])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|