Spaces:
Sleeping
Sleeping
File size: 1,949 Bytes
a124d29 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 | """One-off: load a CSV via the real reader, run the detector with
use_model=True, and print:
- the new categorised method names + per-row attribution,
- the aggregate `model_feature_drivers` map from get_statistics()
(the shape the frontend will chart).
"""
import sys
from pathlib import Path
sys.path.insert(0, str(Path(__file__).parent))
from telecom_anomaly.core.detector import TelecomAnomalyDetector
from telecom_anomaly.core.models import TelecomConfig
CSV = (
r"C:\my_project\work\GH_Police\main project\backend"
r"\telecom_anomaly_detection\data\strictly_cleaned_data"
r"\591296805 calls (1)_strictly_cleaned.csv"
)
cfg = TelecomConfig()
det = TelecomAnomalyDetector(cfg)
det.load_data([CSV])
det.detect_anomalies(use_model=True)
stats = det.get_statistics()
print(f"\nTotal anomalies: {stats.total_anomalies_detected}")
print(f"Rows with anomalies: {stats.rows_with_anomalies}")
print(f"\nanomaly_types (rule + model categories combined):")
for k, v in sorted(stats.anomaly_types.items(), key=lambda x: -x[1]):
print(f" {v:4d} {k}")
print(f"\nmodel_feature_drivers (top-driver counts across all AI flags):")
for k, v in sorted(stats.model_feature_drivers.items(), key=lambda x: -x[1]):
print(f" {v:4d} {k}")
print(f"\nSample AI-flagged rows with category + reason + structured attribution:")
shown = 0
for orig_idx, methods in det.row_anomaly_map.items():
ai = [(i, m) for i, m in enumerate(methods) if m.startswith('AI:')]
if not ai or shown >= 5:
continue
i, method = ai[0]
reason = det.row_anomaly_reasons[orig_idx][i]
attribution = det.row_anomaly_attribution[orig_idx][i]
row = det.original_data[orig_idx]
print(f"\n row {orig_idx}: {row.get('event_date')} {row.get('event_time')} "
f"dur={row.get('duration')}")
print(f" method : {method}")
print(f" reason : {reason}")
print(f" attribution: {attribution}")
shown += 1
|