← Source tree
reviewqueue.py
"""Offline review queue. No records are corrected, removed, or sent anywhere."""
import argparse
import copy
import hashlib
import json
import math
import re
from collections import Counter, defaultdict
from decimal import Decimal, InvalidOperation
from pathlib import Path
from sklearn.ensemble import IsolationForest
FEATURES = ["signed_log_amount", "log_quantity", "signed_log_unit_amount"]
MAX_ROWS = 5000
MIN_REFERENCE = 20
def load_payload(raw):
"""Reject ambiguous/lossy JSON before any source values can be replaced."""
def unique_object(pairs):
result = {}
for key, value in pairs:
if key in result:
raise ValueError(f"duplicate JSON key: {key!r}")
result[key] = value
return result
def safe_float(token):
parsed = float(token)
if not math.isfinite(parsed) or Decimal(token) != Decimal(str(parsed)):
raise ValueError("JSON numeric precision would be lost; use quoted decimal strings")
return parsed
def invalid_constant(token):
raise ValueError(f"nonstandard JSON number: {token}")
return json.loads(raw, object_pairs_hook=unique_object, parse_float=safe_float,
parse_constant=invalid_constant)
def number(value):
"""Accept explicit decimal notation only; never guess locale separators."""
if isinstance(value, bool) or not isinstance(value, (str, int, float)):
raise ValueError("missing_or_invalid_number")
token = str(value).strip()
if not re.fullmatch(r"[+-]?\d+(?:\.\d+)?", token):
raise ValueError("ambiguous_or_invalid_number")
try:
n = Decimal(token)
except InvalidOperation as e:
raise ValueError("invalid_number") from e
if not n.is_finite() or abs(n) > Decimal("1000000000000000"):
raise ValueError("unsupported_numeric_range")
return n
def features(row):
amount, quantity = number(row.get("amount")), number(row.get("quantity"))
if quantity <= 0:
raise ValueError("quantity_must_be_positive")
def signed_log(n):
v = float(n)
if not math.isfinite(v):
raise ValueError("unsupported_derived_numeric_range")
return math.copysign(math.log1p(abs(v)), v)
return [signed_log(amount), math.log1p(float(quantity)), signed_log(amount / quantity)]
def currency(row):
c = row.get("currency")
return c if isinstance(c, str) and re.fullmatch(r"[A-Z]{3}", c) else None
def validate_rows(rows, label):
if not isinstance(rows, list) or len(rows) > MAX_ROWS:
raise ValueError(f"{label}: expected a list of at most {MAX_ROWS} rows")
if any(not isinstance(r, dict) for r in rows):
raise ValueError(f"{label}: every row must be an object")
def review(payload):
if not isinstance(payload, dict):
raise ValueError("input must be an object")
rows, reference = payload.get("records"), payload.get("reference")
validate_rows(rows, "records")
validate_rows(reference, "reference")
groups = defaultdict(list)
for i, row in enumerate(reference):
c = currency(row)
if c is None:
raise ValueError(f"reference row {i + 1}: invalid currency")
try:
groups[c].append(features(row))
except ValueError as e:
raise ValueError(f"reference row {i + 1}: {e}") from e
models, metadata, bounds = {}, {}, {}
for c, x in groups.items():
enough = len(x) >= MIN_REFERENCE and len({tuple(v) for v in x}) >= 5
metadata[c] = {"reference_rows": len(x), "status": "trained" if enough else "insufficient_reference_or_variation"}
if enough:
models[c] = IsolationForest(n_estimators=100, max_samples=min(len(x), 256),
contamination="auto", random_state=42, n_jobs=1).fit(x)
# Promote a model hint only outside observed numeric support. This
# reduces queue noise but can miss low-density interior anomalies.
bounds[c] = [(min(col), max(col)) for col in zip(*x)]
metadata[c].update(flag_policy="model_outlier_and_outside_reference_support",
feature_bounds=[list(pair) for pair in bounds[c]],
policy_notice="Conservative review policy; can miss interior anomalies. Not a calibrated probability.")
ids = Counter(r.get("record_id") for r in rows
if isinstance(r.get("record_id"), str) and r["record_id"].strip())
results = []
for index, row in enumerate(rows, 1):
flags = []
rid = row.get("record_id")
if not isinstance(rid, str) or not rid.strip():
flags.append("missing_or_nontext_record_id")
elif ids[rid] > 1:
flags.append("repeated_record_id")
c = currency(row)
if c is None:
flags.append("missing_or_invalid_currency")
ml = {"status": "not_scored"}
try:
x = features(row)
except ValueError as e:
flags.append(str(e))
ml["reason"] = "invalid_numeric_fields"
else:
if c in models:
score = float(models[c].decision_function([x])[0])
ml = {"status": "scored", "decision_score": round(score, 8),
"reference_currency": c, "model_candidate": score < 0,
"outside_reference_support": any(v < lo or v > hi for v, (lo, hi) in zip(x, bounds[c]))}
if score < 0 and ml["outside_reference_support"]:
flags.append("model_unusual_for_reference")
else:
ml["reason"] = "no_sufficient_same_currency_reference"
results.append({"row_number": index, "original": copy.deepcopy(row),
"flags": flags, "ml": ml, "needs_review": bool(flags)})
return {"tool": "ReviewQueue 0.6", "model": "scikit-learn IsolationForest",
"features": FEATURES, "reference_models": metadata,
"input_rows": len(rows), "output_rows": len(results),
"review_rows": sum(r["needs_review"] for r in results),
"notice": "Review hints only. Scores are not probabilities. No fraud, accounting, legal, or payment decision is made.",
"rows": results}
def main():
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("input", type=Path)
parser.add_argument("--output", type=Path, required=True)
args = parser.parse_args()
if args.input.resolve() == args.output.resolve():
parser.error("input and output must be different")
raw = args.input.read_bytes()
if len(raw) > 8_000_000:
parser.error("input exceeds the 8 MB demonstration limit")
try:
payload = load_payload(raw)
result = review(payload)
result["source_sha256"] = hashlib.sha256(raw).hexdigest()
# Validate UTF-8 before creating a result file (e.g. unpaired surrogates).
encoded = (json.dumps(result, ensure_ascii=False, indent=2, allow_nan=False) + "\n").encode("utf-8")
with args.output.open("xb") as f:
f.write(encoded)
except (ValueError, FileExistsError, RecursionError, OverflowError) as e:
parser.error(str(e))
print(f"Preserved {result['output_rows']} rows; flagged {result['review_rows']} for review.")
if __name__ == "__main__":
main()