chris/OSVauco
1
0
Fork 0
Code Issues 1 Pull Requests Actions 2 Packages Projects Releases Wiki Activity

feat(cg5): anomaly_detector reads BigQuery daily costs

Browse Source
This commit is contained in:
Chris Christiansen 2026-05-27 17:25:44 +00:00
parent 91b71ef3cd
commit 63f3b404b6
1 changed files with 29 additions and 27 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

56
ml/anomaly_detector.py
View File

@ -10,43 +10,45 @@ class AnomalyDetector:
if not self.project_id: if not self.project_id:
raise ValueError("GOOGLE_CLOUD_PROJECT environment variable not set.") raise ValueError("GOOGLE_CLOUD_PROJECT environment variable not set.")
self.billing_table = os.environ.get("BILLING_TABLE")
if not self.billing_table:
raise ValueError("BILLING_TABLE environment variable not set.")
self.pubsub_topic_name = os.environ.get("PUBSUB_TOPIC", "billing_alerts") self.pubsub_topic_name = os.environ.get("PUBSUB_TOPIC", "billing_alerts")
self.bq_client = bigquery.Client(project=self.project_id) self.bq_client = bigquery.Client(project=self.project_id)
self.publisher = pubsub_v1.PublisherClient() self.publisher = pubsub_v1.PublisherClient()
self.topic_path = self.publisher.topic_path(self.project_id, self.pubsub_topic_name) 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): def detect_anomalies(self):
""" """
Detects anomalies in billing data. Detects anomalies in billing data by comparing today's cost to the 7-day average.
""" """
query = f""" query = f"""
WITH WITH daily_costs AS (
cost_last_7_days AS ( SELECT
SELECT service.description AS service,
service.description as service, DATE(_PARTITIONTIME) AS usage_date,
SUM(cost) as total_cost SUM(cost) AS daily_cost
FROM `{self.billing_table}` FROM `{self.billing_table}`
WHERE _PARTITIONTIME >= TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL 8 DAY) WHERE _PARTITIONTIME >= TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL 14 DAY)
AND _PARTITIONTIME < TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL 1 DAY) GROUP BY 1, 2
GROUP BY 1
), ),
cost_last_1_day AS ( costs_with_avg AS (
SELECT SELECT
service.description as service, service,
SUM(cost) as total_cost usage_date,
FROM `{self.billing_table}` daily_cost,
WHERE _PARTITIONTIME >= TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL 1 DAY) AVG(daily_cost) OVER (PARTITION BY service ORDER BY usage_date ROWS BETWEEN 7 PRECEDING AND 1 PRECEDING) AS avg_7day
GROUP BY 1 FROM daily_costs
) )
SELECT SELECT
c1.service, service,
c1.total_cost as today_cost, daily_cost AS today_cost,
c7.total_cost / 7 as avg_7day, avg_7day,
c1.total_cost / (c7.total_cost / 7) as ratio daily_cost / avg_7day AS ratio
FROM cost_last_1_day c1 FROM costs_with_avg
JOIN cost_last_7_days c7 ON c1.service = c7.service WHERE usage_date = CURRENT_DATE()
WHERE c1.total_cost > 2 * (c7.total_cost / 7) AND daily_cost > 2.0 * avg_7day
""" """
query_job = self.bq_client.query(query) query_job = self.bq_client.query(query)
results = query_job.result() results = query_job.result()
@ -56,7 +58,7 @@ class AnomalyDetector:
anomalies.append({ anomalies.append({
"service": row.service, "service": row.service,
"today_cost": row.today_cost, "today_cost": row.today_cost,
"avg_7day": row.avg_7day, "avg_7d": row.avg_7day,
"ratio": row.ratio "ratio": row.ratio
}) })
@ -73,7 +75,7 @@ class AnomalyDetector:
message_data = { message_data = {
"service": anomaly["service"], "service": anomaly["service"],
"today_cost": anomaly["today_cost"], "today_cost": anomaly["today_cost"],
"avg_7day": anomaly["avg_7day"], "avg_7d": anomaly["avg_7d"],
"ratio": anomaly["ratio"], "ratio": anomaly["ratio"],
"timestamp": datetime.datetime.now().isoformat() "timestamp": datetime.datetime.now().isoformat()
} }