feat: add proxy/anomaly detection scoring + retry analysis for failed runs

analysis/entity_aggregator.py

@@ -385,6 +385,36 @@ def aggregate_by_entity(
                 .alias(col)
             )
 
+    # ── Proxy / Anomaly Detection Scores ──────────────────────────────
+    # 1. Non-standard port ratio (ports other than 443, 80, 53, 8080, 8443)
+    port_col = _find_column(schema, (':prd', 'dst_port', '8prd', 'port'))
+    if port_col:
+        std_ports = [443, 80, 53, 8080, 8443]
+        lf = lf.with_columns(
+            (~pl.col(port_col).cast(pl.Utf8).str.replace(r'\.0$', '').cast(pl.Int64, strict=False)
+             .is_in(std_ports)).cast(pl.Float64)
+            .alias('_non_std_port_ratio')
+        )
+
+    # 2. Modern TLS version ratio (TLS 1.2 / 1.3 = 0303 / 0304)
+    tls_col = _find_column(schema, ('0ver', 'tls_version', '_tlsver'))
+    if tls_col:
+        lf = lf.with_columns(
+            pl.col(tls_col).cast(pl.Utf8).str.replace(' ', '')
+            .is_in(['0303', '0304']).cast(pl.Float64)
+            .alias('_modern_tls_ratio')
+        )
+
+    # 3. SNI missing ratio (null or empty string)
+    sni_col = _find_column(schema, _SNI_KEYWORDS)
+    if sni_col:
+        lf = lf.with_columns(
+            pl.col(sni_col).cast(pl.Utf8)
+            .map_batches(lambda s: (s == '') | s.is_null(), return_dtype=pl.Boolean)
+            .cast(pl.Float64)
+            .alias('_sni_missing_ratio')
+        )
+
     # Build aggregation expressions based on the schema (and optional type_map)
     agg_params = build_aggregation_params(schema, type_map=type_map)
 
@@ -396,6 +426,16 @@ def aggregate_by_entity(
         if not name.startswith('_'):
             feature_names.append(name)
 
+    # Add mean aggregation for scoring pre-computed columns
+    _lf_cols = lf.collect_schema().names()
+    for col_name, alias_name in [
+        ('_non_std_port_ratio', 'non_std_port_rate'),
+        ('_modern_tls_ratio', 'modern_tls_rate'),
+    ]:
+        if col_name in _lf_cols:
+            exprs.append(pl.col(col_name).mean().alias(alias_name))
+            feature_names.append(alias_name)
+
     # Check entity columns exist
     missing = [c for c in entity_columns if c not in schema]
     if missing:
@@ -407,6 +447,25 @@ def aggregate_by_entity(
     # Perform aggregation
     aggregated_lf = lf.group_by(entity_columns).agg(exprs)
 
+    # ── Combined proxy_score (post-aggregation) ───────────────────────
+    _agg_cols = aggregated_lf.collect_schema().names()
+    score_parts: list[pl.Expr] = []
+    if 'non_std_port_rate' in _agg_cols:
+        score_parts.append(pl.col('non_std_port_rate') * 0.3)
+    if 'modern_tls_rate' in _agg_cols:
+        score_parts.append((1.0 - pl.col('modern_tls_rate')) * 0.2)
+    if 'sni_missing_ratio' in _agg_cols:
+        score_parts.append(pl.col('sni_missing_ratio') * 0.2)
+    if 'countries_visited' in _agg_cols:
+        score_parts.append((pl.col('countries_visited') > 1).cast(pl.Float64) * 0.3)
+
+    if score_parts:
+        proxy_expr = score_parts[0]
+        for part in score_parts[1:]:
+            proxy_expr = proxy_expr + part
+        aggregated_lf = aggregated_lf.with_columns(proxy_expr.alias('proxy_score'))
+        feature_names.append('proxy_score')
+
     # Post-process: compute active_hours_count if a timestamp column exists
     # (We compute this as a second pass since it requires extraction logic)
     any(name == '_has_timestamp' for name, _ in agg_params)

analysis/urls.py

@@ -12,6 +12,7 @@ urlpatterns = [
     path('analyze/llm/', views.run_llm_analysis_view, name='run_llm_analysis'),
     path('runs/', views.run_list, name='run_list'),
     path('runs/<int:run_id>/', views.run_detail, name='run_detail'),
+    path('runs/<int:run_id>/retry/', views.retry_analysis, name='retry_analysis'),
     path('runs/<int:run_id>/status/', views.run_status_api, name='run_status'),
     path('clusters/<int:run_id>/', views.cluster_overview, name='cluster_overview'),
     path('clusters/<int:run_id>/<int:cluster_label>/', views.cluster_detail, name='cluster_detail'),

analysis/views.py

@@ -43,6 +43,74 @@ def run_detail(request, run_id):
     })
 
 
+@csrf_exempt
+def retry_analysis(request, run_id):
+    """Retry a failed analysis run. Creates a new run with same parameters."""
+    if request.method != 'POST':
+        return JsonResponse({'error': 'POST required'}, status=405)
+
+    original = get_object_or_404(AnalysisRun, id=run_id)
+    if original.status != 'failed':
+        return JsonResponse({'error': '只能重试失败的分析'}, status=400)
+
+    # Create new run copying original parameters
+    new_run = AnalysisRun.objects.create(
+        csv_glob=original.csv_glob,
+        entity_column=original.entity_column,
+        status='pending',
+        total_flows=original.total_flows,
+    )
+
+    def _retry_fn(run, ctx):
+        from analysis.data_loader import load_csv_directory
+        from analysis.session_store import SessionStore
+        from analysis.entity_detector import detect_entity_column
+        from analysis.entity_aggregator import aggregate_by_entity
+
+        store = SessionStore()
+        try:
+            lf, schema, row_count, file_count, memory_mb = load_csv_directory(run.csv_glob)
+            ds_id = f'retry_{run.id}'
+            store.store_dataset(ds_id, lf, schema=schema, metadata={
+                'row_count': row_count, 'file_count': file_count, 'csv_glob': run.csv_glob,
+            })
+            run.total_flows = row_count
+            run.save(update_fields=['total_flows'])
+
+            # Detect entity column
+            result = detect_entity_column(ds_id)
+            entity_col = result.get('recommended')
+            if not entity_col:
+                raise Exception('Could not detect entity column')
+
+            run.entity_column = entity_col
+            run.save(update_fields=['entity_column'])
+
+            # Aggregate
+            agg_lf, feature_columns = aggregate_by_entity(lf, entity_col, schema)
+            df_agg = agg_lf.collect(streaming=True)
+            entity_ds_id = f'entity_{run.id}'
+            store.store_dataset(entity_ds_id, agg_lf, schema=dict(zip(feature_columns, ['']*len(feature_columns))))
+            run.entity_count = len(df_agg)
+            run.save(update_fields=['entity_count'])
+
+            # Run clustering pipeline (same defaults as manual analysis)
+            _run_clustering_pipeline(
+                run=run, store=store, entity_ds_id=entity_ds_id,
+                feature_columns=None, algorithm='hdbscan',
+                min_cluster_size=5, run_pca=True,
+            )
+        except Exception as e:
+            run.status = 'failed'
+            run.error_message = str(e)
+            run.save(update_fields=['status', 'error_message'])
+
+    t = threading.Thread(target=_run_pipeline_worker, args=(new_run.id, _retry_fn), daemon=True)
+    t.start()
+
+    return JsonResponse({'status': 'started', 'run_id': new_run.id, 'redirect': f'/runs/{new_run.id}/'})
+
+
 def _extract_lat(val):
     """Try to extract a latitude from *val* (number or str)."""
     if val is None: