import os import datetime from google.cloud import bigquery class BillingAgent: 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.billing_table = os.environ.get("BILLING_TABLE") if not self.billing_table: raise ValueError("BILLING_TABLE environment variable not set.") self.bq_client = bigquery.Client(project=self.project_id) # ── summary ──────────────────────────────────────────────────────────────────── def get_summary(self): query = f""" SELECT DATE(usage_start_time) AS usage_date, project.id AS project_id, service.description AS service, SUM(cost) AS daily_cost FROM `{self.billing_table}` WHERE _PARTITIONTIME >= TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL 30 DAY) GROUP BY usage_date, project_id, service ORDER BY usage_date DESC, daily_cost DESC LIMIT 100 """ results = self.bq_client.query(query).result() summary = [ { "usage_date": str(row.usage_date), "project_id": row.project_id, "service": row.service, "daily_cost": row.daily_cost, } for row in results ] if not summary: return { "onboarding_status": { "state": "awaiting_data", "message": "Fakturaeksport er aktiv, men ingen data for siste 30 dager ennå.", } } return {"summary": summary} # ── forecast ────────────────────────────────────────────────────────────────── def get_forecast(self): q7 = f""" SELECT SUM(cost) + SUM(IFNULL( (SELECT SUM(c.amount) FROM UNNEST(credits) c), 0)) AS total_cost FROM `{self.billing_table}` WHERE _PARTITIONTIME >= TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL 7 DAY) """ total_7d = list(self.bq_client.query(q7).result())[0].total_cost or 0 daily_average = total_7d / 7 today = datetime.date.today() next_month = datetime.date( today.year + (1 if today.month == 12 else 0), (today.month % 12) + 1, 1 ) remaining_days = (next_month - today).days q_mtd = f""" SELECT SUM(cost) + SUM(IFNULL( (SELECT SUM(c.amount) FROM UNNEST(credits) c), 0)) AS total_cost FROM `{self.billing_table}` WHERE EXTRACT(MONTH FROM _PARTITIONTIME) = EXTRACT(MONTH FROM CURRENT_DATE()) AND EXTRACT(YEAR FROM _PARTITIONTIME) = EXTRACT(YEAR FROM CURRENT_DATE()) """ mtd_cost = list(self.bq_client.query(q_mtd).result())[0].total_cost or 0 return { "daily_average_last_7_days": daily_average, "month_to_date_cost": mtd_cost, "forecasted_remaining_cost": daily_average * remaining_days, "total_monthly_forecast": mtd_cost + daily_average * remaining_days, "remaining_days_in_month": remaining_days, "data_note": "Prognose basert på siste 7 dager. BigQuery kan ha 24-48 timers forsinkelse.", } # ── credits-status (ny) ─────────────────────────────────────────────────────── def get_credits_status(self, days: int = 90): """ CG4-credits: Henter faktiske kreditter fra BigQuery billing export. Returnerer: - credits_used_total: total kreditter brukt hittil (alle typer) - credits_by_type: [{type, full_name, amount}] sortert størst først - gross_cost_total: bruttokostnad uten kreditter - net_cost_total: nettokostnad etter kreditter - daily_gross_burn: gjennomsnittlig daglig bruttokostnad (7d) - daily_credit_burn: gjennomsnittlig daglig kredittforbruk (7d) - credit_runway_days: estimert antall dager til credits er tom - credit_exhaustion_date: estimert dato når credits går tom - data_as_of: siste dato med data i BQ (24-48t forsinkelse) - warning: settes hvis runway < 30 dager """ # 1. Total kreditter brukt og type-breakdown q_credits = f""" SELECT cr.type AS credit_type, cr.full_name AS full_name, ROUND(SUM(cr.amount), 4) AS total_amount FROM `{self.billing_table}`, UNNEST(credits) AS cr WHERE _PARTITIONTIME >= TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL {int(days)} DAY) GROUP BY credit_type, full_name ORDER BY total_amount ASC """ # 2. Daglig burn siste 7 dager (brutto og kreditter separat) q_burn = f""" SELECT ROUND(SUM(cost) / 7, 6) AS daily_gross, ROUND(SUM(IFNULL( (SELECT SUM(c.amount) FROM UNNEST(credits) c), 0)) / 7, 6) AS daily_credit FROM `{self.billing_table}` WHERE _PARTITIONTIME >= TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL 7 DAY) """ # 3. Total brutto + netto hittil q_totals = f""" SELECT ROUND(SUM(cost), 4) AS gross_total, ROUND(SUM(cost) + SUM(IFNULL( (SELECT SUM(c.amount) FROM UNNEST(credits) c), 0)), 4) AS net_total FROM `{self.billing_table}` WHERE _PARTITIONTIME >= TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL {int(days)} DAY) """ # 4. Siste dato med data q_latest = f""" SELECT MAX(DATE(usage_start_time)) AS latest_date FROM `{self.billing_table}` """ credits_rows = list(self.bq_client.query(q_credits).result()) burn_row = list(self.bq_client.query(q_burn).result())[0] totals_row = list(self.bq_client.query(q_totals).result())[0] latest_row = list(self.bq_client.query(q_latest).result())[0] credits_by_type = [ { "type": row.credit_type, "full_name": row.full_name, "amount": float(row.total_amount), } for row in credits_rows ] credits_used_total = abs(sum(r["amount"] for r in credits_by_type)) gross_total = float(totals_row.gross_total or 0) net_total = float(totals_row.net_total or 0) daily_gross = float(burn_row.daily_gross or 0) daily_credit = abs(float(burn_row.daily_credit or 0)) data_as_of = str(latest_row.latest_date) if latest_row.latest_date else None # Runway: Google Vertex AI free tier er typisk $300 USD per prosjekt # Vi beregner gjenstående basert på faktisk brukt vs antatt total-kreditt # Brukeren må sette GOOGLE_CREDIT_TOTAL_USD i env for nøyaktig beregning credit_total_usd = float(os.environ.get("GOOGLE_CREDIT_TOTAL_USD", "0")) runway_days = None exhaustion_date = None credits_remaining = None if credit_total_usd > 0 and daily_credit > 0: credits_remaining = round(credit_total_usd - credits_used_total, 2) runway_days = int(credits_remaining / daily_credit) if credits_remaining > 0 else 0 exhaustion_date = str( datetime.date.today() + datetime.timedelta(days=runway_days) ) if runway_days > 0 else str(datetime.date.today()) elif daily_credit > 0: # Ingen total satt: gi burn rate men ikke runway credits_remaining = None runway_days = None exhaustion_date = None warning = None if runway_days is not None and runway_days < 30: warning = f"⚠️ Kreditter estimert tom om {runway_days} dager ({exhaustion_date}). Aktiver fakturering!" elif runway_days is not None and runway_days < 60: warning = f"⚠️ Kreditter estimert tom om {runway_days} dager ({exhaustion_date})." return { "credits_used_total_usd": round(credits_used_total, 4), "credits_remaining_usd": credits_remaining, "credit_total_usd": credit_total_usd if credit_total_usd > 0 else None, "gross_cost_total_usd": gross_total, "net_cost_total_usd": net_total, "daily_gross_burn_usd": round(daily_gross, 6), "daily_credit_burn_usd": round(daily_credit, 6), "credit_runway_days": runway_days, "credit_exhaustion_date": exhaustion_date, "credits_by_type": credits_by_type, "data_as_of": data_as_of, "period_days": days, "warning": warning, "setup_note": None if credit_total_usd > 0 else ( "Sett GOOGLE_CREDIT_TOTAL_USD i Cloud Run env for nøyaktig runway-beregning. " "Eksempel: 300 for $300 Google gratis-kreditter." ), } # ── anomalier ────────────────────────────────────────────────────────────────── def get_anomalies(self): query = f""" WITH daily_costs AS ( SELECT service.description AS service, DATE(_PARTITIONTIME) AS usage_date, SUM(cost) + SUM(IFNULL( (SELECT SUM(c.amount) FROM UNNEST(credits) c), 0)) AS daily_cost FROM `{self.billing_table}` WHERE _PARTITIONTIME >= TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL 14 DAY) GROUP BY 1, 2 ), costs_with_avg AS ( SELECT service, usage_date, daily_cost, AVG(daily_cost) OVER ( PARTITION BY service ORDER BY usage_date ROWS BETWEEN 7 PRECEDING AND 1 PRECEDING ) AS avg_7day FROM daily_costs ) SELECT service, daily_cost AS today_cost, avg_7day, (daily_cost / avg_7day) AS ratio FROM costs_with_avg WHERE usage_date = CURRENT_DATE() AND avg_7day > 0 AND daily_cost > (2.0 * avg_7day) """ results = self.bq_client.query(query).result() return { "anomalies": [ { "service": row.service, "today_cost": row.today_cost, "avg_7d": row.avg_7day, "ratio": row.ratio, } for row in results ] } # ── daglig historikk for bar-chart (CG6) ─────────────────────────────────── def get_daily_history(self, days: int = 30): query = f""" SELECT DATE(usage_start_time) AS usage_date, SUM(cost) + SUM(IFNULL( (SELECT SUM(c.amount) FROM UNNEST(credits) c), 0)) AS day_cost FROM `{self.billing_table}` WHERE _PARTITIONTIME >= TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL {int(days)} DAY) GROUP BY usage_date ORDER BY usage_date ASC """ results = list(self.bq_client.query(query).result()) history = [] mtd = 0.0 for row in results: if history and row.usage_date.day == 1: mtd = 0.0 mtd += float(row.day_cost or 0) history.append({"date": str(row.usage_date), "mtd": round(mtd, 6)}) return history # ── tjenester gruppert per service (CG5+CG6) ───────────────────────────────── def get_service_totals(self, days: int = 30): query = f""" SELECT service.description AS service, sku.description AS sku, SUM(cost) + SUM(IFNULL( (SELECT SUM(c.amount) FROM UNNEST(credits) c), 0)) AS sku_cost FROM `{self.billing_table}` WHERE _PARTITIONTIME >= TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL {int(days)} DAY) GROUP BY service, sku HAVING sku_cost > 0 ORDER BY service, sku_cost DESC """ results = self.bq_client.query(query).result() services: dict = {} for row in results: svc = row.service if svc not in services: services[svc] = {"service": svc, "total_cost": 0.0, "skus": []} services[svc]["total_cost"] = round( services[svc]["total_cost"] + float(row.sku_cost), 6 ) services[svc]["skus"].append({ "sku": row.sku, "sku_cost": round(float(row.sku_cost), 6), }) return sorted(services.values(), key=lambda x: x["total_cost"], reverse=True) if __name__ == '__main__': agent = BillingAgent() print("Summary:", agent.get_summary()) print("Forecast:", agent.get_forecast()) print("Credits status:", agent.get_credits_status()) print("Anomalies:", agent.get_anomalies()) print("By-service:", agent.get_service_totals()) print("History:", agent.get_daily_history())