TIRTH BHATT

import requests, json, os
from azure.storage.blob import BlobServiceClient
from datetime import datetime

def ingest_weather(api_key: str, city: str, conn_str: str) -> dict:
    # Pull live weather data from Weatherstack API
    resp = requests.get(f"http://api.weatherstack.com/current",
        params={"access_key": api_key, "query": city})
    payload = resp.json()

    # Stage raw JSON to Azure Blob — immutable audit trail
    blob_name = f"raw/weather_{city}_{datetime.utcnow():%Y%m%d_%H%M%S}.json"
    client = BlobServiceClient.from_connection_string(conn_str)
    client.get_blob_client(container="weather-raw", blob=blob_name).upload_blob(json.dumps(payload), overwrite=True)

    return {"status": "staged", "blob": blob_name, "records": 1}

from xgboost import XGBClassifier
from sklearn.model_selection import StratifiedKFold, cross_val_score
import pandas as pd

def build_churn_model(gold_df: pd.DataFrame) -> XGBClassifier:
    # Feature set: recency, frequency, monetary + engagement signals
    features = ["days_since_last_order", "order_frequency",
                "avg_order_value", "support_tickets", "promo_response_rate"]
    X, y = gold_df[features], gold_df["churned"]

    model = XGBClassifier(
        n_estimators=300, max_depth=5, learning_rate=0.05,
        scale_pos_weight=3,   # handle class imbalance
        use_label_encoder=False, eval_metric="auc"
    )
    cv = StratifiedKFold(n_splits=5, shuffle=True, random_state=42)
    scores = cross_val_score(model, X, y, cv=cv, scoring="roc_auc")
    print(f"AUC: {scores.mean():.3f} ± {scores.std():.3f}")

    model.fit(X, y)
    return model

from prophet import Prophet
import pandas as pd

def forecast_demand(df: pd.DataFrame, horizon: int = 90) -> pd.DataFrame:
    # df must have columns: ds (date), y (demand units)
    model = Prophet(
        yearly_seasonality=True,
        weekly_seasonality=True,
        changepoint_prior_scale=0.15,   # flexibility of trend shifts
        interval_width=0.95             # 95% confidence bands
    )
    model.add_country_holidays(country_name="US")
    model.fit(df)

    future  = model.make_future_dataframe(periods=horizon)
    forecast = model.predict(future)

    # Flag reorder: when lower CI drops below safety stock threshold
    forecast["reorder_alert"] = forecast["yhat_lower"] < df["safety_stock"].mean()
    return forecast[["ds", "yhat", "yhat_lower", "yhat_upper", "reorder_alert"]]

// Detect fraud rings: accounts sharing 2+ transaction hubs
// within 48h — a pattern invisible to standard SQL joins
MATCH (a1:Account)-[:SENT_TO]->(hub:Account)
      <-[:SENT_TO]-(a2:Account)
WHERE  a1 <> a2
  AND  hub.total_received > 50000
  AND  hub.account_age_days < 30   // newly opened — red flag
WITH   a1, a2, collect(hub) AS shared_hubs
WHERE  size(shared_hubs) >= 2      // coordinated routing
RETURN a1.account_id, a2.account_id,
       size(shared_hubs)        AS shared_hub_count,
       sum([h IN shared_hubs | h.total_received]) AS total_exposure
ORDER BY total_exposure DESC
LIMIT  50

from dataclasses import dataclass
from typing import Dict, List
import pandas as pd

@dataclass
class SchemaDriftReport:
    added_cols:   List[str]
    dropped_cols: List[str]
    type_changes: Dict[str, tuple]
    def is_clean(self) -> bool:
        return not any([self.added_cols, self.dropped_cols, self.type_changes])

def detect_schema_drift(
    baseline: pd.DataFrame,
    incoming: pd.DataFrame
) -> SchemaDriftReport:
    b_schema = dict(baseline.dtypes)
    i_schema = dict(incoming.dtypes)

    return SchemaDriftReport(
        added_cols   = [c for c in i_schema if c not in b_schema],
        dropped_cols = [c for c in b_schema if c not in i_schema],
        type_changes = {
            c: (str(b_schema[c]), str(i_schema[c]))
            for c in b_schema
            if c in i_schema and b_schema[c] != i_schema[c]
        }
    )

-- Rolling 3-match batting average with YoY comparison
-- Pattern reused in financial rolling KPIs & insurance loss ratios

-- Rolling 5-match average runs (window function via RANKX)
Rolling_5_Avg_Runs =
VAR CurrentMatchRank =
    RANKX( ALL(Matches[MatchID]), Matches[MatchDate],, ASC )
RETURN
AVERAGEX(
    FILTER(
        ALL(Matches),
        RANKX( ALL(Matches[MatchID]), Matches[MatchDate],, ASC )
            >= CurrentMatchRank - 4
        &&
        RANKX( ALL(Matches[MatchID]), Matches[MatchDate],, ASC )
            <= CurrentMatchRank
    ),
    [Total_Runs]
)

-- Season-over-Season growth using SAMEPERIODLASTYEAR
Runs_SoS_Growth_% =
VAR CurrentSeason = [Total_Runs]
VAR PriorSeason   =
    CALCULATE(
        [Total_Runs],
        SAMEPERIODLASTYEAR( Dates[Date] )
    )
RETURN
    IF(
        ISBLANK(PriorSeason) || PriorSeason = 0,
        BLANK(),
        DIVIDE( CurrentSeason - PriorSeason, PriorSeason )
    )

def gate_1_validate(row: dict) -> tuple:
    # phone_1 is the only mandatory field — Bronze stays raw,
    # everything else is nullable and gets cleaned in Silver
    phone = normalize_phone(row.get("phone_1"))

    if not phone or len(phone) != 10:
        return "QUARANTINE", "missing_or_invalid_phone"

    row["phone_1"] = phone
    return "BRONZE", None