cbazza/AFC-Demo

Fork 0

Files

T

cbazza 976af0f2e6 Build functions/ jobs for data_pool and transformation

2025-11-17 15:27:21 +01:00

11 KiB

Raw Blame History

Stufe 2: Transformation vom Data Pool in produktive Tabellen

Aktueller Stand (Stufe 1 ✅)

backend.transactions (74 Transaktionen)
       +
backend.transaction_outputs (7,471 Outputs)
       ↓
       [SyncBackendDataPool Job]
       ↓
public.backend_data_pool (7,471 Datensätze)

Status: ✅ Datenpool ist befüllt und wird alle 6h aktualisiert

Ziel von Stufe 2

public.backend_data_pool (7,471 Datensätze)
       ↓
       [TransformDataPoolToProduction Job] ← NEU
       ↓
public.companies (Unique Companies mit KYC Risk Level)
       +
public.transactions (Transaktionen mit allen 139 Spalten befüllt)

🎯 Was Stufe 2 macht

1. Companies erstellen/aktualisieren

Aus dem Data Pool werden unique Companies extrahiert:

Input (Data Pool):

transaction_id | corporate_entity           | output_key        | content
---------------|----------------------------|-------------------|------------------
1              | Mercedes-Benz Group AG     | corporate_summary | {"answer": "..."}
1              | Mercedes-Benz Group AG     | corporate_sector  | {"answer": "..."}
2              | Samsung Electronics        | corporate_summary | {"answer": "..."}

Output (public.companies):

id | name                    | sector        | country | kyc_risk_level | summary
---|-------------------------|---------------|---------|----------------|----------
1  | Mercedes-Benz Group AG  | Automotive    | DE      | low            | German...
2  | Samsung Electronics     | Electronics   | KR      | high           | Korean...

Wie:

Gruppiere nach corporate_entity (eindeutige Firmennamen)
Extrahiere Company-Daten aus corporate_* outputs
Berechne KYC Risk Level basierend auf:
- tranx_score
- corporate_eusanctions, corporate_ofacsanctions
- country_risk
- corporate_pepexposure, corporate_AMLexposure
- corruption_* outputs

2. Transactions erstellen/aktualisieren

Für jede Transaction im Data Pool wird ein Datensatz in public.transactions erstellt:

Input (Data Pool - gruppiert nach transaction_id):

transaction_id: 1
  - corporate_entity: "Mercedes-Benz Group AG"
  - tx_amount: 54880.9
  - tx_date: "2025-10-28"
  - outputs:
    - corporate_summary: {...}
    - corporate_history: {...}
    - tranx_score: {"score": 45}
    - ... (101 output_keys total)

Output (public.transactions):

id | company_id | reference        | amount   | executed_at | risk_score | corporate_summary | tranx_score | ... (139 Spalten)
---|------------|------------------|----------|-------------|------------|-------------------|-------------|----
1  | 1          | MIGRATED-1       | 54880.9  | 2025-10-28  | 45         | {"answer": "..."} | {"score":45}| ...

Mapping:

Core Felder:
- company_id ← Lookup/Create Company by corporate_entity
- reference ← 'MIGRATED-' || transaction_id
- amount ← tx_amount
- currency ← tx_currency
- executed_at ← tx_date::timestamp
- counterparty ← corporate_counterparty
- status ← status
Output Felder (102 JSONB Spalten):
- corporate_summary ← content WHERE output_key='corporate_summary'
- corporate_history ← content WHERE output_key='corporate_history'
- tranx_score ← content WHERE output_key='tranx_score'
- ... für alle 102 output_keys

🔧 Technische Details

Job-Struktur

class TransformDataPoolToProduction implements ShouldQueue
{
    public function handle()
    {
        // 1. Hole alle unique transactions aus data pool
        $transactions = DB::table('backend_data_pool')
            ->select('transaction_id')
            ->distinct()
            ->get();

        foreach ($transactions as $transaction) {
            // 2. Hole alle Outputs für diese Transaction
            $outputs = $this->getOutputsForTransaction($transaction->transaction_id);

            // 3. Erstelle/Update Company
            $company = $this->createOrUpdateCompany($outputs);

            // 4. Erstelle/Update Transaction
            $this->createOrUpdateTransaction($company, $outputs);
        }
    }

    private function createOrUpdateCompany($outputs)
    {
        $corporateEntity = $outputs->first()->corporate_entity;

        // Berechne KYC Risk Level
        $kycRiskLevel = $this->calculateKycRiskLevel($outputs);

        return Company::updateOrCreate(
            ['name' => $corporateEntity],
            [
                'sector' => $this->extractSector($outputs),
                'country' => $this->extractCountry($outputs),
                'kyc_risk_level' => $kycRiskLevel,
                'summary' => $this->extractSummary($outputs),
                // ... weitere Felder
            ]
        );
    }

    private function createOrUpdateTransaction($company, $outputs)
    {
        $first = $outputs->first();

        // Core Felder
        $transactionData = [
            'company_id' => $company->id,
            'reference' => 'MIGRATED-' . $first->transaction_id,
            'amount' => $first->tx_amount,
            'currency' => $first->tx_currency,
            'executed_at' => Carbon::parse($first->tx_date),
            'status' => $first->status,
            // ...
        ];

        // Output Felder (102 JSONB columns)
        foreach ($outputs as $output) {
            $columnName = $output->output_key;
            $transactionData[$columnName] = json_decode($output->content, true);
        }

        return Transaction::updateOrCreate(
            ['reference' => $transactionData['reference']],
            $transactionData
        );
    }
}

📊 Beispiel-Transformation

Input: Data Pool Datensätze für Transaction ID=1

transaction_id=1, corporate_entity="Mercedes-Benz Group AG", tx_amount=54880.9
  output_key="corporate_summary"  → content={"answer": "..."}
  output_key="corporate_sector"   → content={"answer": "Automotive"}
  output_key="corporate_HQ"       → content={"answer": "Stuttgart, Germany"}
  output_key="tranx_score"        → content={"score": 45}
  output_key="corporate_eusanctions" → content={"found": false}
  ... (101 outputs total)

Output 1: Companies Tabelle

INSERT INTO public.companies (name, sector, headquarters, kyc_risk_level, summary)
VALUES (
  'Mercedes-Benz Group AG',
  'Automotive',
  'Stuttgart, Germany',
  'low',  -- Berechnet aus tranx_score=45, keine Sanctions, etc.
  '...'
);

Output 2: Transactions Tabelle

INSERT INTO public.transactions (
  company_id,
  reference,
  amount,
  currency,
  executed_at,
  status,
  corporate_summary,
  corporate_sector,
  tranx_score,
  corporate_eusanctions,
  ... -- alle 102 output columns
)
VALUES (
  1,  -- Company ID
  'MIGRATED-1',
  54880.9,
  'EUR',
  '2025-10-28',
  'done',
  '{"answer": "..."}',
  '{"answer": "Automotive"}',
  '{"score": 45}',
  '{"found": false}',
  ...
);

🎯 Features von Stufe 2

✅ Intelligente Company-Erstellung

Deduplizierung: Gleicher Name = gleiche Company
Enrichment: Automatische Extraktion von Sector, HQ, etc.
KYC Risk Berechnung: Automatische Risikobewertung

✅ Vollständige Transaction-Daten

Alle 139 Spalten werden befüllt
JSON-Daten aus Outputs werden korrekt gemappt
Referenz-Nummern für Tracking

✅ Idempotent

Mehrfaches Ausführen ist sicher
updateOrCreate() verhindert Duplikate
Bestehende Daten werden aktualisiert

✅ Batch Processing

Verarbeitet Daten in Batches
Kann in Queue laufen
Progress Tracking

🔄 Workflow nach Implementation

1. Backend Daten ändern sich
   ↓
2. SyncBackendDataPool läuft (alle 6h)
   → backend_data_pool aktualisiert
   ↓
3. TransformDataPoolToProduction läuft (nach Sync)
   → Companies aktualisiert
   → Transactions aktualisiert
   ↓
4. Frontend zeigt aktuelle Daten

❓ Offene Fragen für Stufe 2

1. Update-Strategie

Sollen bestehende Transactions überschrieben werden?
Oder nur neue Transactions hinzufügen?

2. Company Matching

Nur exakter Name-Match?
Oder fuzzy matching (z.B. "BASF SE" vs "BASF")?

3. Risk Level Defaults

Was wenn keine Risk-Daten vorhanden?
Default zu "high" (vorsichtig) oder "low"?

4. Fehlende Felder

Was wenn ein output_key fehlt?
NULL speichern oder Default-Wert?

5. Scheduling

Soll Stufe 2 direkt nach Stufe 1 laufen?
Oder separater Schedule?

🔑 KYC Risk Level Berechnung

Verfügbare Risk-Daten

Aus transaction_outputs:

tranx_score - Haupt-Risiko-Score (JSON mit numerischem Wert)
tranx_reasoning - Begründung für den Score
country_risk - Länder-Risiko
sanctions_circumvention - Sanktionsumgehung
corruption_sector - Korruption im Sektor
corruption_country - Korruption im Land
corruption_relationship - Korruption in Beziehungen
corporate_eusanctions - EU Sanktionen
corporate_ofacsanctions - OFAC Sanktionen
corporate_uksanctions - UK Sanktionen
corporate_pepexposure - PEP Exposure
corporate_AMLexposure - AML Exposure
corporate_adverse - Adverse Media

Berechnungs-Logik (3 Risk Levels)

Risk Levels:

low (Geringes Risiko): Score 0-40
high (Hohes Risiko): Score 41-70
critical (Kritisches Risiko): Score 71-100

Gewichtung:

Transaction Score: 40%
Sanctions: 25%
Country Risk: 15%
PEP/Adverse: 10%
Corruption: 10%

Beispiel:

Transaction Score: 45 × 0.40 = 18
Sanctions:        0  × 0.25 = 0
Country Risk:     30 × 0.15 = 4.5
PEP/Adverse:      0  × 0.10 = 0
Corruption:       0  × 0.10 = 0
--------------------------------
Total Score:                 22.5 → "low"

Company-Level Risk:

Aggregiert über alle Transaktionen einer Company
Worst-Case-Prinzip: Eine critical Transaction → Company ist critical
Wenn >30% high → Company ist critical
Wenn >10% high → Company ist high

📋 Implementierungs-Schritte

Phase 1: Vorbereitung

✅ Data Pool ist befüllt
⬜ KYC Risk Calculator Service erstellen
⬜ Data Extraction Helpers erstellen
⬜ Mapping-Logik definieren

Phase 2: Job Implementation

⬜ TransformDataPoolToProduction Job erstellen
⬜ Company Creation Logic implementieren
⬜ Transaction Creation Logic implementieren
⬜ Error Handling & Logging

Phase 3: Testing

⬜ Unit Tests für Risk Calculator
⬜ Feature Tests für Transformation Job
⬜ Datenintegritäts-Checks

Phase 4: Scheduling

⬜ Schedule konfigurieren
⬜ Queue Setup (optional)
⬜ Monitoring einrichten

Phase 5: Deployment

⬜ Produktions-Test mit echten Daten
⬜ Performance-Optimierung
⬜ Dokumentation finalisieren

📈 Erwartete Ergebnisse

Nach erfolgreicher Implementation von Stufe 2:

Companies:

~60-70 unique Companies (geschätzt aus 74 Transaktionen)
Alle mit KYC Risk Level
Enriched mit Sector, Country, HQ, etc.

Transactions:

74 Transaktionen
Alle 139 Spalten befüllt
Verlinkt mit Companies
Referenz-Nummern für Tracking

Performance:

Erste Transformation: ~30-60 Sekunden
Incremental Updates: ~5-10 Sekunden
Kann parallel zu Stufe 1 laufen

🚀 Nächste Schritte

Beantworten der offenen Fragen
Implementation des TransformDataPoolToProduction Jobs
Tests schreiben und ausführen
Scheduling einrichten
Monitoring & Alerts konfigurieren

Erstellt am: 2025-11-16 Status: PLANUNG - Wartet auf Entscheidungen zu offenen Fragen

11 KiB Raw Blame History Unescape Escape