analytics_export.py

"""VynFi Analytics — pre-built statistical evaluations from DataSynth 2.3."""

import os

import vynfi

client = vynfi.VynFi(api_key=os.environ["VYNFI_API_KEY"])

# ── Find a recent completed job ──────────────────────────────────────────────

jobs = client.jobs.list(status="completed", limit=20)

if not jobs.data:
    print("No completed jobs. Generate some data first.")
    raise SystemExit(1)

# ── Pull pre-built analytics from each job ────────────────────────────────────
#
# DataSynth 2.3+ writes four analytics files into every archive:
#   analytics/benford_analysis.json
#   analytics/amount_distribution.json
#   analytics/process_variant_summary.json
#   analytics/banking_evaluation.json
#
# The /v1/jobs/{id}/analytics endpoint returns all four merged into one response.
# Pre-2.3 jobs return 404.

for job in jobs.data:
    print(f"\n=== Job {job.id} ({job.status}) ===")
    try:
        a = client.jobs.analytics(job.id)
    except vynfi.NotFoundError:
        print("  No analytics (pre-DataSynth 2.3 job)")
        continue

    if a.benford_analysis:
        b = a.benford_analysis
        print(f"  Benford's Law:")
        print(f"    Sample size:  {b.sample_size:,}")
        print(f"    MAD:          {b.mad:.4f}")
        print(f"    Conformity:   {b.conformity}")
        print(f"    Chi-squared:  {b.chi_squared:.2f} (p={b.p_value:.4f})")
        print(f"    Passes:       {b.passes}")

    if a.amount_distribution:
        d = a.amount_distribution
        print(f"  Amount Distribution:")
        print(f"    Sample size:  {d.sample_size:,}")
        print(f"    Mean:         {d.mean}")
        print(f"    Median:       {d.median}")
        print(f"    Skewness:     {d.skewness:+.3f}")
        print(f"    Kurtosis:     {d.kurtosis:+.3f}")
        print(f"    Round number ratio: {d.round_number_ratio:.2%}")
        if d.fitted_mu is not None:
            print(f"    Log-normal fit: mu={d.fitted_mu:.2f}, sigma={d.fitted_sigma:.2f}")

    if a.process_variant_summary:
        v = a.process_variant_summary
        print(f"  Process Variants:")
        print(f"    Variant count:    {v.variant_count}")
        print(f"    Total cases:      {v.total_cases}")
        print(f"    Entropy:          {v.variant_entropy:.3f}")
        print(f"    Happy path:       {v.happy_path_concentration:.2%}")
        print(f"    Top variants:")
        for variant_id, freq in v.top_variants[:3]:
            print(f"      {variant_id}: {freq:.2%}")

    if a.banking_evaluation:
        be = a.banking_evaluation
        print(f"  Banking Evaluation:")
        print(f"    Overall passes: {be.passes}")
        if be.kyc:
            print(f"    KYC core fields: {be.kyc.core_field_rate:.2%}")
            print(f"    KYC verification: {be.kyc.verification_rate:.2%}")
        if be.aml:
            print(f"    AML typology coverage: {be.aml.typology_coverage:.2%}")
        if be.cross_layer:
            print(f"    Cross-layer fraud propagation: {be.cross_layer.fraud_propagation_rate:.2%}")
        if be.velocity:
            print(f"    Velocity coverage: {be.velocity.coverage_rate:.2%}")
        if be.false_positive:
            print(f"    AML false positive rate: {be.false_positive.fp_rate:.2%}")
        if be.issues:
            print(f"    Issues: {be.issues}")
    break  # Only show the first job with analytics