# 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):** ```sql 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):** ```sql 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 ```php 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 ```sql 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 ```sql 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:** 1. **low** (Geringes Risiko): Score 0-40 2. **high** (Hohes Risiko): Score 41-70 3. **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 ✅ ABGESCHLOSSEN 1. ✅ Data Pool ist befüllt 2. ✅ KYC Risk Calculator Service erstellen 3. ✅ Data Extraction Helpers erstellen 4. ✅ Mapping-Logik definieren ### Phase 2: Job Implementation ✅ ABGESCHLOSSEN 1. ✅ TransformDataPoolToProduction Job erstellen 2. ✅ Company Creation Logic implementieren 3. ✅ Transaction Creation Logic implementieren 4. ✅ Error Handling & Logging ### Phase 3: Testing 1. ⬜ Unit Tests für Risk Calculator 2. ⬜ Feature Tests für Transformation Job 3. ⬜ Datenintegritäts-Checks ### Phase 4: Scheduling ✅ ABGESCHLOSSEN 1. ✅ Schedule konfigurieren (läuft alle 6h um :30) 2. ✅ Queue Setup (optional) 3. ⬜ Monitoring einrichten ### Phase 5: Deployment 1. ✅ Produktions-Test mit echten Daten 2. ⬜ Performance-Optimierung 3. ⬜ 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 1. **Beantworten der offenen Fragen** 2. **Implementation des TransformDataPoolToProduction Jobs** 3. **Tests schreiben und ausführen** 4. **Scheduling einrichten** 5. **Monitoring & Alerts konfigurieren** --- ## 🎉 Status Update **Erstellt am:** 2025-11-16 **Aktualisiert am:** 2025-11-21 **Status:** ✅ **PRODUKTIV** - Stufe 2 ist implementiert und läuft automatisch! ### Was funktioniert: ✅ **TransformDataPoolToProduction Job** - Vollständig implementiert ✅ **Automatisches Scheduling** - Läuft alle 6 Stunden um :30 (0:30, 6:30, 12:30, 18:30) ✅ **Company & Transaction Creation** - Automatische Erstellung und Updates ✅ **Risk Score Mapping** - Verwendet Backend risk_score direkt ✅ **Artisan Command** - `php artisan backend:transform-data-pool` verfügbar ### Offene TODOs: ⬜ Unit Tests für Transformation Job (Phase 3) ⬜ Feature Tests für Transformation Job (Phase 3) ⬜ Monitoring einrichten (Phase 4) ⬜ Performance-Optimierung (Phase 5) ⬜ KycRiskCalculator Service integrieren (optional, aktuell nicht genutzt)