⚡ pytrackio

from pytrackio import track, timer, counter, report
import time, random

@track
def load_dataset(path: str) -> list:
    time.sleep(0.12)
    counter("rows_ingested").increment(50_000)
    return list(range(50_000))

@track
def engineer_features(data: list) -> list:
    time.sleep(0.09)  # often the slowest step
    return [x / len(data) for x in data]

@track(name="model_training")
def train_model(features: list, labels: list) -> dict:
    time.sleep(0.45)
    counter("models_trained").increment()
    return {"accuracy": 0.923, "f1_score": 0.918}

with timer("end_to_end_pipeline"):
    raw = load_dataset("/data/train.csv")
    features = engineer_features(raw)
    metrics = train_model(features, labels)

report()
# model_training is ~45% of total pipeline time

from fastapi import FastAPI
from pytrackio import track, counter, get_registry

app = FastAPI()

# Use sync helpers — @track doesn't support async def yet
@track
def _get_user_logic(user_id: int) -> dict:
    counter("user_reads").increment()
    return {"id": user_id, "name": f"User-{user_id}"}

@app.get("/users/{user_id}")
async def get_user(user_id: int):
    return _get_user_logic(user_id)

@app.get("/metrics")
def get_metrics():
    # Expose live metrics — add auth before production!
    registry = get_registry()
    return {
        "uptime_seconds": registry.uptime_seconds(),
        "functions": [
            {"name": s.name, "calls": s.calls,
             "avg_ms": round(s.avg_ms, 2),
             "error_rate": round(s.error_rate, 2)}
            for s in registry.all_summaries()
        ]
    }
# uvicorn fastapi_app:app --reload

from pytrackio import track, timer, counter, report
import time, random

@track
def load_train_csv(path: str) -> dict:
    time.sleep(0.08)
    counter("train_rows").increment(10_000)
    return {"shape": (10_000, 47)}

@track
def feature_engineering(train, test) -> tuple:
    time.sleep(0.15)  # often the slowest step
    return train, test

@track(name="xgboost_cv")
def cross_validate(data, folds: int = 5) -> float:
    time.sleep(0.6 * folds / 5)
    counter("cv_folds_run").increment(folds)
    return 0.82 + random.uniform(-0.02, 0.02)

BASE = "/kaggle/input/competition"
with timer("full_competition_pipeline"):
    train = load_train_csv(f"{BASE}/train.csv")
    test = load_train_csv(f"{BASE}/test.csv")
    train, test = feature_engineering(train, test)
    score = cross_validate(train, folds=5)
    print(f"CV Score: {score:.4f}")

report()
# Immediately see: feature_engineering + xgboost_cv are bottlenecks

from pytrackio import track, get_registry, counter
import threading, time, random

ERROR_RATE_THRESHOLD = 5.0   # percent
LATENCY_THRESHOLD_MS = 500   # milliseconds

def continuous_monitor() -> None:
    registry = get_registry()
    while True:
        for s in registry.all_summaries():
            if s.error_rate > ERROR_RATE_THRESHOLD:
                print(f"[CRITICAL] {s.name}: {s.error_rate:.1f}% errors")
            if s.avg_ms > LATENCY_THRESHOLD_MS:
                print(f"[WARNING] {s.name}: {s.avg_ms:.0f}ms latency")
        time.sleep(30)

# Start as daemon — dies with the main process
threading.Thread(target=continuous_monitor, daemon=True).start()

@track
def process_payment(txn_id: str, amount: float) -> bool:
    time.sleep(random.uniform(0.05, 0.6))
    if random.random() < 0.08:  # 8% — will trigger alert
        raise RuntimeError("Payment processor timeout")
    return True