diff --git a/STUFE_2_TRANSFORMATION_PLAN.md b/STUFE_2_TRANSFORMATION_PLAN.md new file mode 100644 index 0000000..1a641cb --- /dev/null +++ b/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* diff --git a/app/Console/Commands/RebuildFromDataPoolCommand.php b/app/Console/Commands/RebuildFromDataPoolCommand.php new file mode 100644 index 0000000..fbb66bf --- /dev/null +++ b/app/Console/Commands/RebuildFromDataPoolCommand.php @@ -0,0 +1,30 @@ +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(); + } +} diff --git a/app/Jobs/TransformDataPoolToProduction.php b/app/Jobs/TransformDataPoolToProduction.php new file mode 100644 index 0000000..d80b1a4 --- /dev/null +++ b/app/Jobs/TransformDataPoolToProduction.php @@ -0,0 +1,375 @@ + $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(), + ]; + } +} diff --git a/app/Models/Transaction.php b/app/Models/Transaction.php index 0e986c0..b24a890 100644 --- a/app/Models/Transaction.php +++ b/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 */ - protected $fillable = [ - 'company_id', - 'reference', - 'amount', - 'currency', - 'counterparty', - 'counterparty_country', - 'channel', - 'executed_at', - 'risk_score', - 'status', - 'requires_review', - 'flagged_by', - 'flagged_reason', - 'signals', - ]; + protected $guarded = []; /** - * @var array + * Cached JSONB column casts to avoid repeated schema queries */ - protected $casts = [ - 'executed_at' => 'datetime', - 'requires_review' => 'boolean', - 'signals' => 'array', - ]; + private static ?array $jsonbCasts = null; + + /** + * 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 + */ + 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_FALSE_POSITIVE => __('Fehlalarm'), - default => __('Freigegeben'), + self::STATUS_TRUE_POSITIVE => __('Kritisches Risiko'), + self::STATUS_FALSE_POSITIVE => __('Hohes Risiko'), + self::STATUS_CLEARED => __('Geringes Risiko'), + default => __('Unbekanntes Risiko'), }; } } diff --git a/app/Services/KycRiskCalculator.php b/app/Services/KycRiskCalculator.php new file mode 100644 index 0000000..6b0bb51 --- /dev/null +++ b/app/Services/KycRiskCalculator.php @@ -0,0 +1,366 @@ + 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'; + } +} diff --git a/misc/database_table_schemes/DB Migration Frontend-Backend.key b/misc/database_table_schemes/DB Migration Frontend-Backend.key index 1f68298..b5f9fcc 100644 Binary files a/misc/database_table_schemes/DB Migration Frontend-Backend.key and b/misc/database_table_schemes/DB Migration Frontend-Backend.key differ diff --git a/misc/database_table_schemes/DB Migration Frontend-Backend.pdf b/misc/database_table_schemes/DB Migration Frontend-Backend.pdf index 1bc6c6e..7ed589b 100644 Binary files a/misc/database_table_schemes/DB Migration Frontend-Backend.pdf and b/misc/database_table_schemes/DB Migration Frontend-Backend.pdf differ diff --git a/routes/console.php b/routes/console.php index 0f8ea14..2ef60de 100644 --- a/routes/console.php +++ b/routes/console.php @@ -1,9 +1,9 @@ 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