From c0a248948fb700591aa431d562805f4407b38bba Mon Sep 17 00:00:00 2001
From: Chris Christiansen <chris.christian...@>
Date: Wed, 27 May 2026 16:59:17 +0000
Subject: [PATCH] feat(cg2): add anomaly_detector.py with Pub/Sub alerting

---
 ml/anomaly_detector.py | 85 ++++++++++++++++++++++++++++++++++++++++++
 1 file changed, 85 insertions(+)
 create mode 100644 ml/anomaly_detector.py

diff --git a/ml/anomaly_detector.py b/ml/anomaly_detector.py
new file mode 100644
index 0000000..ba2dc88
--- /dev/null
+++ b/ml/anomaly_detector.py
@@ -0,0 +1,85 @@
+import os
+import json
+import datetime
+from google.cloud import bigquery
+from google.cloud import pubsub_v1
+
+class AnomalyDetector:
+    def __init__(self):
+        self.project_id = os.environ.get("GOOGLE_CLOUD_PROJECT")
+        if not self.project_id:
+            raise ValueError("GOOGLE_CLOUD_PROJECT environment variable not set.")
+        
+        self.pubsub_topic_name = os.environ.get("PUBSUB_TOPIC", "billing_alerts")
+        self.bq_client = bigquery.Client(project=self.project_id)
+        self.publisher = pubsub_v1.PublisherClient()
+        self.topic_path = self.publisher.topic_path(self.project_id, self.pubsub_topic_name)
+        self.billing_table = f"{self.project_id}.billing_export.gcp_billing_export_v1_*"
+
+    def detect_anomalies(self):
+        """
+        Detects anomalies in billing data.
+        """
+        query = f"""
+            WITH
+            cost_last_7_days AS (
+              SELECT
+                service.description as service,
+                SUM(cost) as total_cost
+              FROM `{self.billing_table}`
+              WHERE _PARTITIONTIME >= TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL 8 DAY)
+                AND _PARTITIONTIME < TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL 1 DAY)
+              GROUP BY 1
+            ),
+            cost_last_1_day AS (
+              SELECT
+                service.description as service,
+                SUM(cost) as total_cost
+              FROM `{self.billing_table}`
+              WHERE _PARTITIONTIME >= TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL 1 DAY)
+              GROUP BY 1
+            )
+            SELECT
+              c1.service,
+              c1.total_cost as today_cost,
+              c7.total_cost / 7 as avg_7day,
+              c1.total_cost / (c7.total_cost / 7) as ratio
+            FROM cost_last_1_day c1
+            JOIN cost_last_7_days c7 ON c1.service = c7.service
+            WHERE c1.total_cost > 2 * (c7.total_cost / 7)
+        """
+        query_job = self.bq_client.query(query)
+        results = query_job.result()
+
+        anomalies = []
+        for row in results:
+            anomalies.append({
+                "service": row.service,
+                "today_cost": row.today_cost,
+                "avg_7day": row.avg_7day,
+                "ratio": row.ratio
+            })
+
+        if anomalies:
+            self.publish_alerts(anomalies)
+
+        return {"anomalies": anomalies}
+
+    def publish_alerts(self, anomalies):
+        """
+        Publishes anomaly alerts to a Pub/Sub topic.
+        """
+        for anomaly in anomalies:
+            message_data = {
+                "service": anomaly["service"],
+                "today_cost": anomaly["today_cost"],
+                "avg_7day": anomaly["avg_7day"],
+                "ratio": anomaly["ratio"],
+                "timestamp": datetime.datetime.now().isoformat()
+            }
+            message_bytes = json.dumps(message_data).encode("utf-8")
+            self.publisher.publish(self.topic_path, data=message_bytes)
+
+if __name__ == '__main__':
+    detector = AnomalyDetector()
+    print("Anomalies:", detector.detect_anomalies())