Gemini API × Google Analytics 4: AI-Powered Business Intelligence — Anomaly Detection, Forecasting & Automated Reporting

# Install required Python packages
pip install google-analytics-data google-generativeai python-dotenv pandas matplotlib slack-sdk

GOOGLE_APPLICATION_CREDENTIALS=path/to/service-account-key.json
GA4_PROPERTY_ID=12345678
GEMINI_API_KEY=YOUR_GEMINI_API_KEY
SLACK_BOT_TOKEN=YOUR_SLACK_BOT_TOKEN
SLACK_CHANNEL_ID=YOUR_CHANNEL_ID

import os
from dotenv import load_dotenv
from google.analytics.data_v1beta import BetaAnalyticsDataClient
from google.analytics.data_v1beta.types import (
    DateRange, Dimension, Metric, RunReportRequest
)
import pandas as pd
 
load_dotenv()
PROPERTY_ID = os.environ["GA4_PROPERTY_ID"]
 
def get_ga4_data(start_date: str, end_date: str, dimensions: list, metrics: list) -> pd.DataFrame:
    """Fetch GA4 data and return it as a DataFrame."""
    client = BetaAnalyticsDataClient()
 
    request = RunReportRequest(
        property=f"properties/{PROPERTY_ID}",
        dimensions=[Dimension(name=d) for d in dimensions],
        metrics=[Metric(name=m) for m in metrics],
        date_ranges=[DateRange(start_date=start_date, end_date=end_date)],
    )
 
    response = client.run_report(request)
 
    rows = []
    for row in response.rows:
        row_dict = {}
        for i, dim in enumerate(dimensions):
            row_dict[dim] = row.dimension_values[i].value
        for i, metric in enumerate(metrics):
            row_dict[metric] = float(row.metric_values[i].value)
        rows.append(row_dict)
 
    return pd.DataFrame(rows)
 
# Quick test
df = get_ga4_data(
    start_date="7daysAgo",
    end_date="today",
    dimensions=["date", "sessionSourceMedium"],
    metrics=["sessions", "engagedSessions", "conversions", "totalRevenue"]
)
print(df.head())
# Sample output:
#         date sessionSourceMedium  sessions  engagedSessions  conversions  totalRevenue
# 0  20260401          google / cpc    1243.0          867.0        42.0       125400.0
# 1  20260401       organic / search    3421.0         2156.0        89.0       267000.0

import google.generativeai as genai
import json
 
genai.configure(api_key=os.environ["GEMINI_API_KEY"])
 
# Define the GA4 query function for Gemini to call
ga4_query_tool = genai.protos.Tool(
    function_declarations=[
        genai.protos.FunctionDeclaration(
            name="get_analytics_data",
            description="Fetch analytics data from GA4 using the specified parameters.",
            parameters=genai.protos.Schema(
                type=genai.protos.Type.OBJECT,
                properties={
                    "start_date": genai.protos.Schema(
                        type=genai.protos.Type.STRING,
                        description="Start date (e.g., '7daysAgo', '30daysAgo', '2026-04-01')"
                    ),
                    "end_date": genai.protos.Schema(
                        type=genai.protos.Type.STRING,
                        description="End date (e.g., 'today', 'yesterday', '2026-04-06')"
                    ),
                    "dimensions": genai.protos.Schema(
                        type=genai.protos.Type.ARRAY,
                        items=genai.protos.Schema(type=genai.protos.Type.STRING),
                        description="GA4 dimension names (date, sessionSourceMedium, deviceCategory, country, landingPage, etc.)"
                    ),
                    "metrics": genai.protos.Schema(
                        type=genai.protos.Type.ARRAY,
                        items=genai.protos.Schema(type=genai.protos.Type.STRING),
                        description="GA4 metric names (sessions, engagedSessions, conversions, totalRevenue, bounceRate, etc.)"
                    ),
                },
                required=["start_date", "end_date", "dimensions", "metrics"]
            )
        )
    ]
)
 
SYSTEM_PROMPT = """You are an expert Google Analytics 4 analyst. When given a question,
fetch the relevant GA4 data and provide a clear, actionable analysis.
Include percentage changes, trend observations, likely root causes, and specific recommendations.
Keep the language accessible to non-technical marketing stakeholders."""
 
def natural_language_analytics(question: str) -> str:
    """Query GA4 data using plain English."""
    model = genai.GenerativeModel(
        model_name="gemini-2.5-pro",
        system_instruction=SYSTEM_PROMPT,
        tools=[ga4_query_tool]
    )
 
    chat = model.start_chat()
    response = chat.send_message(question)
 
    # Handle Function Calling loop
    while response.candidates[0].content.parts[0].function_call.name:
        fc = response.candidates[0].content.parts[0].function_call
        func_args = dict(fc.args)
 
        df = get_ga4_data(
            start_date=func_args["start_date"],
            end_date=func_args["end_date"],
            dimensions=func_args["dimensions"],
            metrics=func_args["metrics"]
        )
 
        data_json = df.to_json(orient="records", indent=2)
 
        response = chat.send_message(
            genai.protos.Part(
                function_response=genai.protos.FunctionResponse(
                    name=fc.name,
                    response={"result": data_json, "rows": len(df)}
                )
            )
        )
 
        if not response.candidates[0].content.parts[0].HasField("function_call"):
            break
 
    return response.text
 
# Example usage
result = natural_language_analytics(
    "Compare conversions by source/medium for the last two weeks and highlight the biggest changes."
)
print(result)
# Sample output:
# Week-over-Week Conversion Comparison (Mar 30 – Apr 5 vs Mar 23–29)
#
# Notable Changes:
# - organic / search: +25.8% (89 → 112 conversions)
# - google / cpc: -33.3% (42 → 28 conversions)
#
# Analysis: The CPC drop coincides with a 3-day campaign pause (Mar 31 – Apr 2)...

import numpy as np
from scipy import stats
 
def detect_anomalies(df: pd.DataFrame, metric: str, z_threshold: float = 2.5) -> pd.DataFrame:
    """Detect statistical outliers in a time-series metric using Z-scores."""
    values = df[metric].values
    z_scores = np.abs(stats.zscore(values))
 
    df_result = df.copy()
    df_result["z_score"] = z_scores
    df_result["is_anomaly"] = z_scores > z_threshold
    df_result["deviation_pct"] = (
        (df[metric] - df[metric].mean()) / df[metric].mean() * 100
    ).round(1)
 
    return df_result
 
def analyze_anomalies_with_gemini(anomaly_df: pd.DataFrame, metric: str) -> str:
    """Use Gemini to explain detected anomalies and suggest corrective actions."""
    anomalies = anomaly_df[anomaly_df["is_anomaly"]].copy()
    if anomalies.empty:
        return f"No statistical anomalies detected for {metric} in the analysis period."
 
    summary = {
        "metric": metric,
        "period_avg": round(anomaly_df[metric].mean(), 1),
        "period_std": round(anomaly_df[metric].std(), 1),
        "anomalies": anomalies[["date", metric, "z_score", "deviation_pct"]].to_dict(orient="records")
    }
 
    model = genai.GenerativeModel("gemini-2.5-pro")
    prompt = f"""
Analyze the following GA4 anomaly detection results:
 
Metric: {summary['metric']}
30-day average: {summary['period_avg']}
Standard deviation: {summary['period_std']}
Detected anomalies:
{json.dumps(summary['anomalies'], indent=2)}
 
Please provide:
1. The severity and business impact of each anomaly
2. Most likely root causes (seasonality, campaigns, technical issues, external events)
3. What additional data to investigate
4. Specific, prioritized action items
 
Write for a marketing manager — clear, concise, and actionable.
"""
    return model.generate_content(prompt).text
 
# Run anomaly detection on 30-day session data
df_30days = get_ga4_data(
    "30daysAgo", "today",
    ["date"],
    ["sessions", "conversions", "engagedSessions"]
)
df_30days = df_30days.sort_values("date")
 
anomaly_result = detect_anomalies(df_30days, "sessions")
analysis = analyze_anomalies_with_gemini(anomaly_result, "sessions")
print(analysis)

def compute_derived_metrics(df: pd.DataFrame) -> pd.DataFrame:
    """Add calculated metrics like CVR and engagement rate."""
    df = df.copy()
    df["cvr"] = (df["conversions"] / df["sessions"] * 100).round(3)
    df["engagement_rate"] = (df["engagedSessions"] / df["sessions"] * 100).round(1)
    return df
 
def run_full_anomaly_scan() -> dict:
    """Scan all key metrics for anomalies."""
    df = get_ga4_data("30daysAgo", "today", ["date"],
                       ["sessions", "conversions", "engagedSessions", "totalRevenue"])
    df = df.sort_values("date")
    df = compute_derived_metrics(df)
 
    metrics_to_check = ["sessions", "cvr", "engagement_rate", "totalRevenue"]
    results = {}
 
    for metric in metrics_to_check:
        anomaly_df = detect_anomalies(df, metric)
        anomalies = anomaly_df[anomaly_df["is_anomaly"]]
 
        if not anomalies.empty:
            results[metric] = {
                "anomaly_count": len(anomalies),
                "worst_day": anomalies.loc[anomalies["z_score"].idxmax(), "date"],
                "worst_deviation": anomalies["deviation_pct"].abs().max(),
                "analysis": analyze_anomalies_with_gemini(anomaly_df, metric)
            }
 
    return results
 
scan_results = run_full_anomaly_scan()
for metric, data in scan_results.items():
    print(f"\n{'='*50}")
    print(f"⚠️  {metric} — {data['anomaly_count']} anomalies detected")
    print(f"Worst day: {data['worst_day']} ({data['worst_deviation']:.1f}% deviation)")

from datetime import datetime, timedelta
 
def generate_weekly_report() -> str:
    """Auto-generate a weekly GA4 performance report using Gemini."""
 
    today = datetime.now()
    last_monday = today - timedelta(days=today.weekday() + 7)
    last_sunday = last_monday + timedelta(days=6)
    prev_monday = last_monday - timedelta(days=7)
    prev_sunday = last_monday - timedelta(days=1)
 
    def fmt(d): return d.strftime("%Y-%m-%d")
 
    # Fetch this week and last week by channel
    channel_this = get_ga4_data(fmt(last_monday), fmt(last_sunday),
        ["sessionSourceMedium"], ["sessions", "engagedSessions", "conversions", "totalRevenue"])
    channel_prev = get_ga4_data(fmt(prev_monday), fmt(prev_sunday),
        ["sessionSourceMedium"], ["sessions", "engagedSessions", "conversions", "totalRevenue"])
 
    # Device breakdown
    device_data = get_ga4_data(fmt(last_monday), fmt(last_sunday),
        ["deviceCategory"], ["sessions", "engagedSessions", "conversions"])
 
    # Top landing pages
    page_data = get_ga4_data(fmt(last_monday), fmt(last_sunday),
        ["landingPage"], ["sessions", "engagedSessions", "conversions"])
    page_data = page_data.sort_values("sessions", ascending=False).head(10)
 
    model = genai.GenerativeModel("gemini-2.5-pro")
 
    prompt = f"""
Write a professional weekly GA4 analytics report for stakeholders.
 
This week ({fmt(last_monday)} – {fmt(last_sunday)}) by channel:
{channel_this.to_string(index=False)}
 
Last week ({fmt(prev_monday)} – {fmt(prev_sunday)}) by channel:
{channel_prev.to_string(index=False)}
 
Device breakdown (this week):
{device_data.to_string(index=False)}
 
Top 10 landing pages (this week):
{page_data.to_string(index=False)}
 
Report structure:
1. Executive Summary (3 sentences max)
2. Key KPI comparison vs. prior week (sessions, CVR, revenue)
3. Channel performance highlights and concerns
4. Device trends and mobile optimization status
5. Top landing page insights and optimization opportunities
6. Top 3 recommended actions for next week (with priority)
 
Format in Markdown. Be concise, data-driven, and executive-friendly.
"""
    return model.generate_content(prompt).text
 
weekly_report = generate_weekly_report()
print(weekly_report)

from slack_sdk import WebClient
from slack_sdk.errors import SlackApiError
 
def send_report_to_slack(report_text: str, anomaly_results: dict):
    """Post the weekly report and any anomaly alerts to Slack."""
    client = WebClient(token=os.environ["SLACK_BOT_TOKEN"])
    channel = os.environ["SLACK_CHANNEL_ID"]
 
    try:
        # Send anomaly alerts first if any exist
        if anomaly_results:
            alert_lines = ["*⚠️ Anomaly Alerts This Week*"]
            for metric, data in anomaly_results.items():
                alert_lines.append(
                    f"• `{metric}`: {data['anomaly_count']} anomalies detected "
                    f"(max deviation: {data['worst_deviation']:.1f}% on {data['worst_day']})"
                )
            client.chat_postMessage(channel=channel, text="\n".join(alert_lines))
 
        # Post the weekly report in chunks (Slack's 4000-char limit)
        max_len = 3000
        chunks = [report_text[i:i+max_len] for i in range(0, len(report_text), max_len)]
 
        for i, chunk in enumerate(chunks):
            prefix = "*📊 Weekly GA4 Performance Report*\n\n" if i == 0 else ""
            client.chat_postMessage(channel=channel, text=prefix + chunk)
 
        print("✅ Slack delivery successful")
 
    except SlackApiError as e:
        print(f"❌ Slack error: {e.response['error']}")
 
# Run the full pipeline
anomalies = run_full_anomaly_scan()
report = generate_weekly_report()
send_report_to_slack(report, anomalies)

from google.cloud import bigquery
 
def natural_language_to_bigquery(nl_question: str, project_id: str, dataset: str) -> tuple:
    """Convert a plain-English question into a BigQuery SQL query and execute it."""
 
    schema_description = f"""
GA4 BigQuery export table: `{project_id}.{dataset}.events_*`
Key columns:
- event_date (STRING): Date in YYYYMMDD format
- event_timestamp (INTEGER): Microsecond timestamp
- event_name (STRING): Event type (page_view, purchase, scroll, etc.)
- user_pseudo_id (STRING): Anonymous user identifier
- geo.country (STRING): User country
- device.category (STRING): Device type (mobile/desktop/tablet)
- traffic_source.medium (STRING): Traffic medium
- ecommerce.purchase_revenue (FLOAT): Purchase revenue
- event_params: Nested key-value array of event parameters
"""
 
    model = genai.GenerativeModel("gemini-2.5-pro")
 
    prompt = f"""
{schema_description}
 
Write a BigQuery SQL query to answer the following question.
Question: {nl_question}
 
Requirements:
- Use the last 30 days of data with: _TABLE_SUFFIX BETWEEN FORMAT_DATE('%Y%m%d', DATE_SUB(CURRENT_DATE(), INTERVAL 30 DAY)) AND FORMAT_DATE('%Y%m%d', CURRENT_DATE())
- Return only executable standard SQL (no explanations or markdown)
"""
 
    response = model.generate_content(prompt)
    sql = response.text.strip().replace("```sql", "").replace("```", "").strip()
 
    bq_client = bigquery.Client(project=project_id)
    df = bq_client.query(sql).to_dataframe()
 
    return df, sql
 
# Example
df_result, generated_sql = natural_language_to_bigquery(
    "Which are the top 5 landing pages by purchase conversion rate for mobile users?",
    project_id="my-analytics-project",
    dataset="analytics_12345678"
)
print("Generated SQL:", generated_sql[:300])
print("Results:", df_result)

# main.py — Cloud Functions entry point
import functions_framework
from flask import Request
 
@functions_framework.http
def weekly_analytics_report(request: Request):
    """Weekly reporting endpoint for Cloud Functions."""
    try:
        anomaly_results = run_full_anomaly_scan()
        report = generate_weekly_report()
        send_report_to_slack(report, anomaly_results)
 
        print(f"Report completed: {datetime.now().isoformat()}")
        print(f"Anomalies detected: {len(anomaly_results)}")
 
        return {"status": "success", "anomaly_count": len(anomaly_results)}, 200
 
    except Exception as e:
        print(f"Error: {str(e)}")
        return {"status": "error", "message": str(e)}, 500

gcloud scheduler jobs create http weekly-analytics-report \
  --schedule="0 0 * * MON" \
  --uri="https://YOUR_CLOUD_FUNCTION_URL" \
  --time-zone="Asia/Tokyo" \
  --message-body="{}" \
  --headers="Content-Type=application/json"

# crontab -e
0 9 * * 1 cd /path/to/project && python run_weekly_report.py >> /var/log/analytics.log 2>&1

def analyze_user_segments() -> str:
    """Analyze performance across new vs. returning users by device and country."""
 
    segment_data = get_ga4_data(
        "30daysAgo", "today",
        ["newVsReturning", "deviceCategory", "country"],
        ["sessions", "engagedSessions", "conversions", "totalRevenue", "userEngagementDuration"]
    )
 
    segment_data["avg_engagement_min"] = (
        segment_data["userEngagementDuration"] / segment_data["sessions"] / 60
    ).round(1)
 
    model = genai.GenerativeModel("gemini-2.5-pro")
 
    prompt = f"""
Analyze the following GA4 user segment data:
 
{segment_data.to_string(index=False)}
 
Please provide:
1. Key differences between new and returning user behavior
2. Which segments convert best and why
3. Correlation between engagement time and conversion rate
4. Country and device combinations worth prioritizing
5. Three high-impact optimization recommendations with expected outcomes
 
Be specific and data-driven. Write for a marketing strategist.
"""
    return model.generate_content(prompt).text
 
segment_analysis = analyze_user_segments()
print(segment_analysis)