Build functions/ jobs for data_pool and transformation

STUFE_2_TRANSFORMATION_PLAN.md

@@ -0,0 +1,428 @@
+# 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
+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
+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
+1. ⬜ Schedule konfigurieren
+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**
+
+---
+
+*Erstellt am: 2025-11-16*
+*Status: PLANUNG - Wartet auf Entscheidungen zu offenen Fragen*

app/Console/Commands/RebuildFromDataPoolCommand.php

@@ -0,0 +1,30 @@
+<?php
+
+namespace App\Console\Commands;
+
+use Illuminate\Console\Command;
+
+class RebuildFromDataPoolCommand extends Command
+{
+    /**
+     * The name and signature of the console command.
+     *
+     * @var string
+     */
+    protected $signature = 'app:rebuild-from-data-pool-command';
+
+    /**
+     * The console command description.
+     *
+     * @var string
+     */
+    protected $description = 'Command description';
+
+    /**
+     * Execute the console command.
+     */
+    public function handle()
+    {
+        //
+    }
+}

app/Console/Commands/TransformDataPoolCommand.php

@@ -0,0 +1,144 @@
+<?php
+
+declare(strict_types=1);
+
+namespace App\Console\Commands;
+
+use App\Jobs\TransformDataPoolToProduction;
+use Illuminate\Console\Command;
+
+use function Laravel\Prompts\confirm;
+use function Laravel\Prompts\info;
+use function Laravel\Prompts\spin;
+use function Laravel\Prompts\table;
+use function Laravel\Prompts\warning;
+
+class TransformDataPoolCommand extends Command
+{
+    /**
+     * The name and signature of the console command.
+     */
+    protected $signature = 'backend:transform-data-pool
+                            {--batch-size=100 : Number of records to process per batch}
+                            {--queue : Dispatch the job to the queue instead of running synchronously}
+                            {--stats : Show statistics only, do not perform transformation}';
+
+    /**
+     * The console command description.
+     */
+    protected $description = 'Transform backend_data_pool into companies and transactions tables';
+
+    /**
+     * Execute the console command.
+     */
+    public function handle(): int
+    {
+        // Show stats only
+        if ($this->option('stats')) {
+            return $this->showStats();
+        }
+
+        info('Preparing for DATA POOL TRANSFORMATION');
+        warning('This will create/update companies and transactions from the data pool.');
+
+        if (! $this->option('no-interaction') && ! confirm('Do you want to continue?', true)) {
+            info('Transformation cancelled.');
+
+            return self::SUCCESS;
+        }
+
+        $batchSize = (int) $this->option('batch-size');
+
+        // Show current stats before transformation
+        $this->showStatsBefore();
+
+        // Create the job
+        $job = new TransformDataPoolToProduction(batchSize: $batchSize);
+
+        // Execute the job
+        if ($this->option('queue')) {
+            info('Dispatching job to queue...');
+            dispatch($job);
+            info('Job dispatched successfully!');
+        } else {
+            info('Running transformation synchronously...');
+
+            spin(
+                fn () => $job->handle(app(\App\Services\KycRiskCalculator::class)),
+                'Transforming data...'
+            );
+
+            info('Transformation completed successfully!');
+        }
+
+        // Show stats after transformation
+        $this->showStatsAfter();
+
+        return self::SUCCESS;
+    }
+
+    /**
+     * Show statistics only
+     */
+    private function showStats(): int
+    {
+        $job = new TransformDataPoolToProduction();
+        $stats = $job->getStats();
+
+        info('Data Pool Transformation Statistics');
+
+        table(
+            ['Metric', 'Value'],
+            [
+                ['Data Pool Transactions', number_format($stats['data_pool_transactions'])],
+                ['Companies in Database', number_format($stats['companies_count'])],
+                ['Transactions in Database', number_format($stats['transactions_count'])],
+                ['Migrated Transactions', number_format($stats['migrated_transactions'])],
+            ]
+        );
+
+        return self::SUCCESS;
+    }
+
+    /**
+     * Show stats before transformation
+     */
+    private function showStatsBefore(): void
+    {
+        $job = new TransformDataPoolToProduction();
+        $stats = $job->getStats();
+
+        $this->newLine();
+        info('Stats BEFORE transformation:');
+        table(
+            ['Metric', 'Value'],
+            [
+                ['Data Pool Transactions', number_format($stats['data_pool_transactions'])],
+                ['Companies in Database', number_format($stats['companies_count'])],
+                ['Migrated Transactions', number_format($stats['migrated_transactions'])],
+            ]
+        );
+        $this->newLine();
+    }
+
+    /**
+     * Show stats after transformation
+     */
+    private function showStatsAfter(): void
+    {
+        $job = new TransformDataPoolToProduction();
+        $stats = $job->getStats();
+
+        $this->newLine();
+        info('Stats AFTER transformation:');
+        table(
+            ['Metric', 'Value'],
+            [
+                ['Data Pool Transactions', number_format($stats['data_pool_transactions'])],
+                ['Companies in Database', number_format($stats['companies_count'])],
+                ['Migrated Transactions', number_format($stats['migrated_transactions'])],
+            ]
+        );
+        $this->newLine();
+    }
+}

app/Jobs/TransformDataPoolToProduction.php

@@ -0,0 +1,375 @@
+<?php
+
+declare(strict_types=1);
+
+namespace App\Jobs;
+
+use App\Models\Company;
+use App\Models\Transaction;
+use App\Services\KycRiskCalculator;
+use Carbon\Carbon;
+use Illuminate\Bus\Queueable;
+use Illuminate\Contracts\Queue\ShouldQueue;
+use Illuminate\Foundation\Bus\Dispatchable;
+use Illuminate\Queue\InteractsWithQueue;
+use Illuminate\Queue\SerializesModels;
+use Illuminate\Support\Facades\DB;
+use Illuminate\Support\Facades\Log;
+
+class TransformDataPoolToProduction implements ShouldQueue
+{
+    use Dispatchable;
+    use InteractsWithQueue;
+    use Queueable;
+    use SerializesModels;
+
+    public int $timeout = 3600; // 1 hour timeout
+
+    public int $tries = 3;
+
+    /**
+     * Create a new job instance.
+     */
+    public function __construct(
+        public ?int $batchSize = 100,
+    ) {
+    }
+
+    /**
+     * Execute the job.
+     */
+    public function handle(KycRiskCalculator $kycCalculator): void
+    {
+        $startTime = now();
+
+        Log::info('TransformDataPoolToProduction: Starting transformation', [
+            'batch_size' => $this->batchSize,
+        ]);
+
+        try {
+            DB::transaction(function () use ($kycCalculator) {
+                $this->transformData($kycCalculator);
+            });
+
+            $duration = now()->diffInSeconds($startTime);
+
+            Log::info('TransformDataPoolToProduction: Transformation completed successfully', [
+                'duration_seconds' => $duration,
+            ]);
+        } catch (\Exception $e) {
+            Log::error('TransformDataPoolToProduction: Transformation failed', [
+                'error' => $e->getMessage(),
+                'trace' => $e->getTraceAsString(),
+            ]);
+
+            throw $e;
+        }
+    }
+
+    /**
+     * Transform data from backend_data_pool to companies and transactions
+     */
+    private function transformData(KycRiskCalculator $kycCalculator): void
+    {
+        // Get all unique transaction IDs from data pool
+        $transactionIds = DB::table('backend_data_pool')
+            ->select('transaction_id')
+            ->distinct()
+            ->pluck('transaction_id');
+
+        Log::info('TransformDataPoolToProduction: Found transactions to process', [
+            'count' => $transactionIds->count(),
+        ]);
+
+        $processedCompanies = 0;
+        $processedTransactions = 0;
+
+        foreach ($transactionIds as $transactionId) {
+            // Get all outputs for this transaction
+            $outputs = DB::table('backend_data_pool')
+                ->where('transaction_id', $transactionId)
+                ->get();
+
+            if ($outputs->isEmpty()) {
+                continue;
+            }
+
+            $firstOutput = $outputs->first();
+
+            // 1. Create or update Company
+            $company = $this->createOrUpdateCompany($firstOutput, $outputs, $kycCalculator);
+            if ($company->wasRecentlyCreated) {
+                $processedCompanies++;
+            }
+
+            // 2. Create or update Transaction
+            $transaction = $this->createOrUpdateTransaction($company, $firstOutput, $outputs);
+            if ($transaction->wasRecentlyCreated) {
+                $processedTransactions++;
+            }
+        }
+
+        Log::info('TransformDataPoolToProduction: Processing completed', [
+            'new_companies' => $processedCompanies,
+            'new_transactions' => $processedTransactions,
+        ]);
+    }
+
+    /**
+     * Create or update a Company from data pool outputs
+     */
+    private function createOrUpdateCompany($firstOutput, $outputs, KycRiskCalculator $kycCalculator): Company
+    {
+        $corporateEntity = $firstOutput->corporate_entity;
+
+        // Calculate KYC Risk Level
+        $kycRiskLevel = $kycCalculator->calculateKycRiskLevel(collect([$outputs])->first());
+
+        // Extract company data from outputs
+        $companyData = [
+            'country' => $this->extractCountry($firstOutput, $outputs),
+            'kyc_risk_level' => $kycRiskLevel,
+            'sector' => $this->extractField($outputs, 'corporate_sector', 255),
+            'headquarters' => $this->extractField($outputs, 'corporate_HQ', 255),
+            'summary' => $this->extractField($outputs, 'corporate_summary', 5000), // text field, longer OK
+            'legal_name' => $this->extractField($outputs, 'corporate_name', 255),
+        ];
+
+        return Company::updateOrCreate(
+            ['name' => $corporateEntity],
+            $companyData
+        );
+    }
+
+    /**
+     * Create or update a Transaction from data pool outputs
+     */
+    private function createOrUpdateTransaction(Company $company, $firstOutput, $outputs): Transaction
+    {
+        $reference = 'MIGRATED-' . $firstOutput->transaction_id;
+
+        // Map KYC Risk Level to Transaction Status
+        $status = match ($company->kyc_risk_level) {
+            'critical' => Transaction::STATUS_TRUE_POSITIVE,
+            'high' => Transaction::STATUS_FALSE_POSITIVE,
+            'low' => Transaction::STATUS_CLEARED,
+            default => Transaction::STATUS_FALSE_POSITIVE, // Default to high risk
+        };
+
+        // Core transaction data
+        $transactionData = [
+            'company_id' => $company->id,
+            'amount' => $firstOutput->tx_amount,
+            'currency' => $firstOutput->tx_currency ?? 'EUR',
+            'counterparty' => $firstOutput->corporate_counterparty,
+            'counterparty_country' => $this->extractCountryCode($firstOutput->tx_country_incoming),
+            'channel' => $firstOutput->source_file,
+            'executed_at' => $this->parseDate($firstOutput->tx_date),
+            'risk_score' => $this->extractRiskScore($outputs),
+            'status' => $status,
+            'requires_review' => true,
+            'flagged_reason' => $firstOutput->tx_purpose,
+        ];
+
+        // Map all output_keys to their respective columns
+        foreach ($outputs as $output) {
+            $columnName = $output->output_key;
+
+            // Skip if column doesn't exist in transactions table
+            if (! $this->columnExists($columnName)) {
+                continue;
+            }
+
+            // Decode JSON content, store NULL if empty or invalid
+            $content = json_decode($output->content, true);
+            $transactionData[$columnName] = $content ?: null;
+        }
+
+        return Transaction::updateOrCreate(
+            ['reference' => $reference],
+            $transactionData
+        );
+    }
+
+    /**
+     * Extract country code from data
+     */
+    private function extractCountry($firstOutput, $outputs): string
+    {
+        // Try to get from tx_country_incoming
+        if (! empty($firstOutput->tx_country_incoming)) {
+            $country = $this->extractCountryCode($firstOutput->tx_country_incoming);
+            if ($country) {
+                return $country;
+            }
+        }
+
+        // Try to extract from corporate_HQ
+        $hq = $this->extractField($outputs, 'corporate_HQ');
+        if ($hq && is_array($hq) && isset($hq['answer'])) {
+            // Simple pattern matching for country codes
+            if (preg_match('/\b([A-Z]{2})\b/', $hq['answer'], $matches)) {
+                return $matches[1];
+            }
+        }
+
+        return 'DE'; // Default
+    }
+
+    /**
+     * Extract country code (convert to ISO 2-letter if needed)
+     */
+    private function extractCountryCode(?string $country): ?string
+    {
+        if (empty($country)) {
+            return null;
+        }
+
+        // If already 2 letters, return uppercase
+        if (strlen($country) === 2) {
+            return strtoupper($country);
+        }
+
+        // Simple country name mapping (extend as needed)
+        $countryMap = [
+            'Germany' => 'DE',
+            'Deutschland' => 'DE',
+            'United States' => 'US',
+            'USA' => 'US',
+            'United Kingdom' => 'GB',
+            'UK' => 'GB',
+            'France' => 'FR',
+            'Spain' => 'ES',
+            'Italy' => 'IT',
+            'Netherlands' => 'NL',
+            'Belgium' => 'BE',
+            'Austria' => 'AT',
+            'Switzerland' => 'CH',
+            'Poland' => 'PL',
+            'Czech Republic' => 'CZ',
+            'Denmark' => 'DK',
+            'Sweden' => 'SE',
+            'Norway' => 'NO',
+            'Finland' => 'FI',
+        ];
+
+        return $countryMap[$country] ?? null;
+    }
+
+    /**
+     * Extract risk score from tranx_score output
+     */
+    private function extractRiskScore($outputs): int
+    {
+        $scoreOutput = $outputs->firstWhere('output_key', 'tranx_score');
+
+        if (! $scoreOutput) {
+            return 128; // Default: Medium risk (255/2)
+        }
+
+        $scoreData = json_decode($scoreOutput->content, true);
+
+        if (is_array($scoreData) && isset($scoreData['score'])) {
+            // Assume score is 0-100, convert to 0-255
+            $score = (int) $scoreData['score'];
+
+            return (int) round(($score / 100) * 255);
+        }
+
+        return 128;
+    }
+
+    /**
+     * Extract a field from outputs
+     */
+    private function extractField($outputs, string $outputKey, int $maxLength = 255): mixed
+    {
+        $output = $outputs->firstWhere('output_key', $outputKey);
+
+        if (! $output) {
+            return null;
+        }
+
+        $data = json_decode($output->content, true);
+
+        // Extract 'answer' field if it exists
+        if (is_array($data) && isset($data['answer'])) {
+            $answer = $data['answer'];
+
+            // If answer is an object/array, convert to string
+            if (is_array($answer)) {
+                $answer = json_encode($answer);
+            }
+
+            // Truncate to max length if needed
+            return $this->truncateString((string) $answer, $maxLength);
+        }
+
+        // If data is an object/array, convert to JSON string
+        if (is_array($data)) {
+            return $this->truncateString(json_encode($data), $maxLength);
+        }
+
+        return $this->truncateString((string) $data, $maxLength);
+    }
+
+    /**
+     * Truncate string to max length
+     */
+    private function truncateString(string $text, int $maxLength): string
+    {
+        if (strlen($text) <= $maxLength) {
+            return $text;
+        }
+
+        return substr($text, 0, $maxLength - 3) . '...';
+    }
+
+    /**
+     * Parse date string to Carbon instance
+     */
+    private function parseDate(?string $date): ?Carbon
+    {
+        if (empty($date)) {
+            return null;
+        }
+
+        try {
+            return Carbon::parse($date);
+        } catch (\Exception $e) {
+            Log::warning('Failed to parse date', ['date' => $date]);
+
+            return null;
+        }
+    }
+
+    /**
+     * Check if column exists in transactions table
+     */
+    private function columnExists(string $columnName): bool
+    {
+        static $columns = null;
+
+        if ($columns === null) {
+            $columns = DB::getSchemaBuilder()->getColumnListing('transactions');
+        }
+
+        return in_array($columnName, $columns);
+    }
+
+    /**
+     * Get statistics about the transformation
+     */
+    public function getStats(): array
+    {
+        return [
+            'data_pool_transactions' => DB::table('backend_data_pool')
+                ->select('transaction_id')
+                ->distinct()
+                ->count(),
+            'companies_count' => Company::count(),
+            'transactions_count' => Transaction::count(),
+            'migrated_transactions' => Transaction::where('reference', 'like', 'MIGRATED-%')->count(),
+        ];
+    }
+}

app/Models/Transaction.php

@@ -17,33 +17,66 @@ class Transaction extends Model
     public const STATUS_CLEARED = 'cleared';
 
     /**
+     * Disable mass assignment protection to allow JSONB columns
+     *
      * @var array<int, string>
      */
-    protected $fillable = [
+    protected $guarded = [];
-        'company_id',
-        'reference',
-        'amount',
-        'currency',
-        'counterparty',
-        'counterparty_country',
-        'channel',
-        'executed_at',
-        'risk_score',
-        'status',
-        'requires_review',
-        'flagged_by',
-        'flagged_reason',
-        'signals',
-    ];
 
     /**
-     * @var array<string, string>
+     * Cached JSONB column casts to avoid repeated schema queries
      */
-    protected $casts = [
+    private static ?array $jsonbCasts = null;
-        'executed_at' => 'datetime',
+
-        'requires_review' => 'boolean',
+    /**
-        'signals' => 'array',
+     * Get the attributes that should be cast.
-    ];
+     *
+     * Dynamically casts all JSONB columns (102 output_keys) as arrays
+     * Uses static caching to avoid repeated database schema queries
+     *
+     * @return array<string, string>
+     */
+    protected function casts(): array
+    {
+        // Return cached casts if available
+        if (self::$jsonbCasts !== null) {
+            return self::$jsonbCasts;
+        }
+
+        $standardCasts = [
+            'executed_at' => 'datetime',
+            'requires_review' => 'boolean',
+            'signals' => 'array',
+        ];
+
+        // Standard non-JSONB columns
+        $standardColumns = [
+            'id', 'company_id', 'reference', 'amount', 'currency',
+            'counterparty', 'counterparty_country', 'channel', 'executed_at',
+            'risk_score', 'status', 'requires_review', 'flagged_by',
+            'flagged_reason', 'signals', 'created_at', 'updated_at',
+        ];
+
+        try {
+            // Get all columns from the database (only once per request)
+            $allColumns = \Illuminate\Support\Facades\Schema::getColumnListing('transactions');
+
+            // All remaining columns are JSONB columns that should be cast as arrays
+            $jsonbColumns = array_diff($allColumns, $standardColumns);
+
+            foreach ($jsonbColumns as $column) {
+                $standardCasts[$column] = 'array';
+            }
+        } catch (\Exception $e) {
+            // If schema query fails (e.g., during migrations), just use standard casts
+            \Illuminate\Support\Facades\Log::warning('Failed to load JSONB column casts: ' . $e->getMessage());
+        }
+
+        // Cache for subsequent calls
+        self::$jsonbCasts = $standardCasts;
+
+        return self::$jsonbCasts;
+    }
 
     public function company(): BelongsTo
     {
@@ -53,9 +86,10 @@ class Transaction extends Model
     public function statusLabel(): string
     {
         return match ($this->status) {
-            self::STATUS_TRUE_POSITIVE => __('Bestätigter Treffer'),
+            self::STATUS_TRUE_POSITIVE => __('Kritisches Risiko'),
-            self::STATUS_FALSE_POSITIVE => __('Fehlalarm'),
+            self::STATUS_FALSE_POSITIVE => __('Hohes Risiko'),
-            default => __('Freigegeben'),
+            self::STATUS_CLEARED => __('Geringes Risiko'),
+            default => __('Unbekanntes Risiko'),
         };
     }
 }

app/Services/KycRiskCalculator.php

@@ -0,0 +1,366 @@
+<?php
+
+declare(strict_types=1);
+
+namespace App\Services;
+
+use Illuminate\Support\Collection;
+
+class KycRiskCalculator
+{
+    /**
+     * KYC Risk Level Schwellenwerte (3 Levels)
+     */
+    private const RISK_THRESHOLDS = [
+        'low' => 40,      // 0-40 = Geringes Risiko
+        'high' => 70,     // 41-70 = Hohes Risiko
+        'critical' => 100, // 71-100 = Kritisches Risiko
+    ];
+
+    /**
+     * Gewichtung der verschiedenen Risk-Faktoren
+     */
+    private const RISK_WEIGHTS = [
+        'transaction_score' => 0.40,  // 40% Gewichtung
+        'sanctions' => 0.25,           // 25% Gewichtung
+        'country_risk' => 0.15,        // 15% Gewichtung
+        'pep_adverse' => 0.10,         // 10% Gewichtung
+        'corruption' => 0.10,          // 10% Gewichtung
+    ];
+
+    /**
+     * Berechne KYC Risk Level basierend auf Transaction Outputs
+     *
+     * @param Collection $outputs Collection von backend_data_pool records für eine Transaction
+     * @return string 'low', 'high', or 'critical'
+     */
+    public function calculateKycRiskLevel(Collection $outputs): string
+    {
+        // Wenn keine Outputs vorhanden: Default = high (vorsichtig)
+        if ($outputs->isEmpty()) {
+            return 'high';
+        }
+
+        // Konvertiere Collection zu Key-Value Array für einfacheren Zugriff
+        $outputsArray = $outputs->pluck('content', 'output_key')->toArray();
+
+        $scores = [
+            'transaction_score' => $this->calculateTransactionScore($outputsArray),
+            'sanctions' => $this->calculateSanctionsScore($outputsArray),
+            'country_risk' => $this->calculateCountryRiskScore($outputsArray),
+            'pep_adverse' => $this->calculatePepAdverseScore($outputsArray),
+            'corruption' => $this->calculateCorruptionScore($outputsArray),
+        ];
+
+        $weightedScore = $this->calculateWeightedScore($scores);
+
+        return $this->determineRiskLevel($weightedScore);
+    }
+
+    /**
+     * Berechne Transaction Score (0-100)
+     */
+    private function calculateTransactionScore(array $outputs): int
+    {
+        if (! isset($outputs['tranx_score'])) {
+            return 60; // Default: Hohes Risiko wenn keine Daten
+        }
+
+        $scoreData = json_decode($outputs['tranx_score'], true);
+
+        // Format: {"score": 75}
+        if (is_array($scoreData) && isset($scoreData['score'])) {
+            return min(100, max(0, (int) $scoreData['score']));
+        }
+
+        // Format: {"risk_level": "high"}
+        if (is_array($scoreData) && isset($scoreData['risk_level'])) {
+            return $this->mapRiskLevelToScore($scoreData['risk_level']);
+        }
+
+        // Falls nur numerischer Wert
+        if (is_numeric($scoreData)) {
+            return min(100, max(0, (int) $scoreData));
+        }
+
+        return 60; // Default
+    }
+
+    /**
+     * Berechne Sanctions Score (0-100)
+     */
+    private function calculateSanctionsScore(array $outputs): int
+    {
+        $sanctionFlags = [
+            'corporate_eusanctions',
+            'corporate_ofacsanctions',
+            'corporate_uksanctions',
+            'sanctions_circumvention',
+        ];
+
+        $score = 0;
+        $foundSanctions = 0;
+
+        foreach ($sanctionFlags as $flag) {
+            if (! isset($outputs[$flag])) {
+                continue;
+            }
+
+            $data = json_decode($outputs[$flag], true);
+
+            if ($this->hasSanctionsMatch($data)) {
+                $foundSanctions++;
+
+                if (is_array($data) && isset($data['severity'])) {
+                    $score += $this->mapSeverityToScore($data['severity']);
+                } else {
+                    $score += 70; // Default: Hohes Risiko
+                }
+            }
+        }
+
+        if ($foundSanctions === 0) {
+            return 0;
+        }
+
+        $avgScore = $score / $foundSanctions;
+
+        // Mehrere Sanctions → Kritisches Risiko
+        if ($foundSanctions > 1) {
+            return min(100, (int) round($avgScore * 1.3));
+        }
+
+        return (int) round($avgScore);
+    }
+
+    /**
+     * Berechne Country Risk Score (0-100)
+     */
+    private function calculateCountryRiskScore(array $outputs): int
+    {
+        if (! isset($outputs['country_risk'])) {
+            return 40; // Default: Geringes Risiko
+        }
+
+        $countryRiskData = json_decode($outputs['country_risk'], true);
+
+        if (is_array($countryRiskData) && isset($countryRiskData['score'])) {
+            return min(100, max(0, (int) $countryRiskData['score']));
+        }
+
+        if (is_array($countryRiskData) && isset($countryRiskData['level'])) {
+            return $this->mapRiskLevelToScore($countryRiskData['level']);
+        }
+
+        if (is_bool($countryRiskData)) {
+            return $countryRiskData ? 80 : 20;
+        }
+
+        return 40;
+    }
+
+    /**
+     * Berechne PEP & Adverse Media Score (0-100)
+     */
+    private function calculatePepAdverseScore(array $outputs): int
+    {
+        $score = 0;
+        $count = 0;
+
+        $pepFields = [
+            'corporate_pepexposure' => 75,
+            'corporate_pep' => 80,
+            'corporate_adverse' => 70,
+            'corporate_AMLexposure' => 85,
+        ];
+
+        foreach ($pepFields as $field => $defaultScore) {
+            if (! isset($outputs[$field])) {
+                continue;
+            }
+
+            $data = json_decode($outputs[$field], true);
+
+            if ($this->hasPositiveMatch($data)) {
+                $score += $this->extractScoreFromData($data, $defaultScore);
+                $count++;
+            }
+        }
+
+        return $count > 0 ? (int) round($score / $count) : 0;
+    }
+
+    /**
+     * Berechne Corruption Score (0-100)
+     */
+    private function calculateCorruptionScore(array $outputs): int
+    {
+        $corruptionFields = [
+            'corruption_sector',
+            'corruption_country',
+            'corruption_relationship',
+            'corporate_corruptionexposure',
+        ];
+
+        $totalScore = 0;
+        $count = 0;
+
+        foreach ($corruptionFields as $field) {
+            if (! isset($outputs[$field])) {
+                continue;
+            }
+
+            $data = json_decode($outputs[$field], true);
+
+            if ($this->hasPositiveMatch($data)) {
+                $totalScore += $this->extractScoreFromData($data, 70);
+                $count++;
+            }
+        }
+
+        return $count > 0 ? (int) round($totalScore / $count) : 0;
+    }
+
+    /**
+     * Berechne gewichteten Gesamt-Score
+     */
+    private function calculateWeightedScore(array $scores): float
+    {
+        $totalScore = 0.0;
+
+        foreach ($scores as $category => $score) {
+            $weight = self::RISK_WEIGHTS[$category] ?? 0;
+            $totalScore += $score * $weight;
+        }
+
+        return $totalScore;
+    }
+
+    /**
+     * Bestimme Risk Level basierend auf Score (3 Levels)
+     */
+    private function determineRiskLevel(float $score): string
+    {
+        if ($score <= self::RISK_THRESHOLDS['low']) {
+            return 'low';  // Geringes Risiko
+        }
+
+        if ($score <= self::RISK_THRESHOLDS['high']) {
+            return 'high';  // Hohes Risiko
+        }
+
+        return 'critical';  // Kritisches Risiko
+    }
+
+    /**
+     * Hilfsfunktionen
+     */
+    private function mapRiskLevelToScore(string $level): int
+    {
+        return match (strtolower($level)) {
+            'low', 'green', 'gering', 'geringes risiko' => 25,
+            'high', 'yellow', 'orange', 'hoch', 'hohes risiko' => 60,
+            'critical', 'red', 'kritisch', 'kritisches risiko' => 85,
+            default => 60,
+        };
+    }
+
+    private function mapSeverityToScore(string $severity): int
+    {
+        return match (strtolower($severity)) {
+            'minor', 'low', 'gering' => 40,
+            'moderate', 'high', 'hoch' => 65,
+            'critical', 'severe', 'kritisch' => 90,
+            default => 65,
+        };
+    }
+
+    private function hasSanctionsMatch($data): bool
+    {
+        if (! is_array($data)) {
+            return false;
+        }
+
+        return (isset($data['matches']) && ! empty($data['matches']))
+            || (isset($data['found']) && $data['found'] === true)
+            || (isset($data['sanctioned']) && $data['sanctioned'] === true);
+    }
+
+    private function hasPositiveMatch($data): bool
+    {
+        if (is_array($data)) {
+            return (isset($data['match']) && $data['match'] === true)
+                || (isset($data['found']) && $data['found'] === true)
+                || (isset($data['matches']) && ! empty($data['matches']))
+                || (isset($data['exposure']) && $data['exposure'] === true);
+        }
+
+        if (is_bool($data)) {
+            return $data;
+        }
+
+        return false;
+    }
+
+    private function extractScoreFromData($data, int $default): int
+    {
+        if (is_array($data) && isset($data['score'])) {
+            return min(100, max(0, (int) $data['score']));
+        }
+
+        if (is_array($data) && isset($data['risk_score'])) {
+            return min(100, max(0, (int) $data['risk_score']));
+        }
+
+        return $default;
+    }
+
+    /**
+     * Berechne aggregierten Risk Level für eine Company basierend auf allen Transaktionen
+     */
+    public function calculateCompanyRiskLevel(Collection $transactions): string
+    {
+        if ($transactions->isEmpty()) {
+            return 'high'; // Default: Hohes Risiko wenn keine Daten
+        }
+
+        $riskScores = [
+            'low' => 0,
+            'high' => 0,
+            'critical' => 0,
+        ];
+
+        // Zähle Risk Levels aller Transaktionen
+        foreach ($transactions as $transactionOutputs) {
+            $riskLevel = $this->calculateKycRiskLevel($transactionOutputs);
+            $riskScores[$riskLevel]++;
+        }
+
+        $totalTransactions = $transactions->count();
+
+        // Worst-Case-Prinzip:
+        // 1. Wenn EINE Transaktion kritisches Risiko hat → Company ist kritisch
+        if ($riskScores['critical'] > 0) {
+            return 'critical';
+        }
+
+        // 2. Wenn >30% der Transaktionen hohes Risiko haben → Company ist kritisch
+        $highRiskPercentage = $riskScores['high'] / $totalTransactions;
+        if ($highRiskPercentage > 0.3) {
+            return 'critical';
+        }
+
+        // 3. Wenn >10% hohes Risiko → Company ist hohes Risiko
+        if ($highRiskPercentage > 0.1) {
+            return 'high';
+        }
+
+        // 4. Wenn >80% geringes Risiko → Company ist geringes Risiko
+        if ($riskScores['low'] / $totalTransactions > 0.8) {
+            return 'low';
+        }
+
+        // 5. Default: Hohes Risiko (vorsichtig)
+        return 'high';
+    }
+}

routes/console.php

@@ -1,9 +1,9 @@
 <?php
 
 use App\Jobs\SyncBackendDataPool;
+use App\Jobs\TransformDataPoolToProduction;
 use Illuminate\Foundation\Inspiring;
 use Illuminate\Support\Facades\Artisan;
-use Illuminate\Support\Facades\Log;
 use Illuminate\Support\Facades\Schedule;
 
 Artisan::command('inspire', function () {
@@ -30,3 +30,17 @@ Schedule::job(new SyncBackendDataPool(fullSync: true, batchSize: 1000))
     ->name('sync-backend-data-pool-full')
     ->description('Full sync and rebuild of backend data pool')
     ->withoutOverlapping(3600); // Max 1 Stunde Lock (3600 Sekunden)
+
+/**
+ * Data Pool Transformation Scheduling (Stufe 2)
+ *
+ * - Läuft 30 Minuten nach jedem Incremental Sync
+ * - Transformiert Daten aus backend_data_pool → companies + transactions
+ */
+
+// Transformation: 30 Minuten nach jedem Sync (0:30, 6:30, 12:30, 18:30)
+Schedule::job(new TransformDataPoolToProduction(batchSize: 100))
+    ->cron('30 */6 * * *') // :30 jede 6 Stunden
+    ->name('transform-data-pool-to-production')
+    ->description('Transform data pool into companies and transactions')
+    ->withoutOverlapping(1800); // Max 30 Minuten Lock