पूर्वानुमान वितरण और विश्वास स्कोर की निगरानी

import datetime
import json
import os
import torch
import torch.nn.functional as F

PREDICTION_LOG = '/tmp/prediction_log.jsonl'

def predict_and_log(features, model, model_version='v1.2.3'):
    import numpy as np
    with torch.no_grad():
        logits = model(torch.tensor(features, dtype=torch.float32).unsqueeze(0))
        probs = F.softmax(logits, dim=1).squeeze().numpy()
    pred_class = int(probs.argmax())
    confidence = float(probs.max())

    record = {
        'timestamp': datetime.datetime.utcnow().isoformat(),
        'model_version': model_version,
        'predicted_class': pred_class,
        'confidence': round(confidence, 4),
        'probabilities': probs.tolist()
    }
    with open(PREDICTION_LOG, 'a') as f:
        f.write(json.dumps(record) + '\n')
    return pred_class, confidence

import numpy as np
import pandas as pd

np.random.seed(42)

# Simulate confidence scores over 12 weeks
# First 6 weeks: healthy (high confidence)
week1to6 = np.random.beta(8, 2, 600)   # mean ~0.8
# Last 6 weeks: degrading (lower confidence)
week7to12 = np.random.beta(3, 3, 600)  # mean ~0.5

all_scores = np.concatenate([week1to6, week7to12])
dates = pd.date_range('2024-01-01', periods=len(all_scores), freq='h')

df = pd.DataFrame({'timestamp': dates, 'confidence': all_scores})
df['week'] = df['timestamp'].dt.isocalendar().week
weekly_mean = df.groupby('week')['confidence'].mean()

threshold = weekly_mean.iloc[:6].quantile(0.05)  # 5th percentile of healthy period
print(f'Alert threshold: {threshold:.4f}')
for week, mean_conf in weekly_mean.items():
    status = 'ALERT' if mean_conf < threshold else 'OK'
    print(f'Week {week}: mean confidence = {mean_conf:.4f} [{status}]')

import numpy as np
from scipy.stats import chi2_contingency

# Baseline class distribution from training predictions
train_predictions = np.array([0, 1]).repeat([700, 300])  # 70% neg, 30% pos
np.random.shuffle(train_predictions)

# Current week production predictions (shifted distribution)
prod_predictions = np.random.choice([0, 1], p=[0.92, 0.08], size=500)

train_counts = [np.sum(train_predictions == 0), np.sum(train_predictions == 1)]
prod_counts  = [np.sum(prod_predictions == 0),  np.sum(prod_predictions == 1)]

chi2, p_val, _, _ = chi2_contingency([train_counts, prod_counts])
print(f'Train distribution: {train_counts[0]/len(train_predictions):.2%} neg, {train_counts[1]/len(train_predictions):.2%} pos')
print(f'Prod  distribution: {prod_counts[0]/len(prod_predictions):.2%} neg, {prod_counts[1]/len(prod_predictions):.2%} pos')
print(f'Chi-squared: {chi2:.2f}, p-value: {p_val:.4f}')
print('Prediction distribution drift:', p_val < 0.05)

import numpy as np
from sklearn.calibration import calibration_curve

np.random.seed(42)
# Overconfident model: high probabilities but not that accurate
y_true = np.random.binomial(1, 0.6, 1000)
y_prob_uncalibrated = np.clip(
    np.random.beta(5, 2, 1000) * (y_true * 0.6 + 0.2), 0, 1
)

fraction_of_positives, mean_predicted = calibration_curve(
    y_true, y_prob_uncalibrated, n_bins=10
)

print('Bin | Predicted | Observed')
for pred, obs in zip(mean_predicted, fraction_of_positives):
    gap = abs(pred - obs)
    status = '*** MISCALIBRATED' if gap > 0.1 else 'OK'
    print(f'{pred:.2f} | {obs:.2f} {status}')

import numpy as np
from sklearn.calibration import calibration_curve

def expected_calibration_error(y_true, y_prob, n_bins=10):
    fraction_pos, mean_pred = calibration_curve(y_true, y_prob, n_bins=n_bins)
    bin_sizes = []
    bins = np.linspace(0, 1, n_bins + 1)
    for i in range(n_bins):
        in_bin = (y_prob >= bins[i]) & (y_prob < bins[i+1])
        bin_sizes.append(in_bin.sum())

    n = len(y_true)
    ece = sum(
        (bin_size / n) * abs(pred - obs)
        for bin_size, pred, obs in zip(bin_sizes[:len(mean_pred)],
                                        mean_pred, fraction_pos)
    )
    return round(ece, 4)

np.random.seed(42)
y_true = np.random.binomial(1, 0.5, 1000)
y_prob = np.random.beta(3, 3, 1000)
ece = expected_calibration_error(y_true, y_prob)
print(f'ECE: {ece}')
print('Calibration quality:', 'Good' if ece < 0.05 else 'Poor')

import datetime

CONFIDENCE_THRESHOLD = 0.75
REVIEW_QUEUE = []

def predict_with_human_review(request_id, features, model_func):
    pred_class, confidence = model_func(features)

    result = {
        'request_id': request_id,
        'predicted_class': pred_class,
        'confidence': confidence,
        'timestamp': datetime.datetime.utcnow().isoformat(),
        'needs_review': confidence < CONFIDENCE_THRESHOLD
    }

    if result['needs_review']:
        REVIEW_QUEUE.append({**result, 'features': features})
        print(f'Request {request_id} FLAGGED for review (conf={confidence:.2f})')
    else:
        print(f'Request {request_id} auto-decided: class={pred_class} (conf={confidence:.2f})')

    return result

import psycopg2
import json
from datetime import datetime

# Schema (run once):
# CREATE TABLE prediction_log (
#     id SERIAL PRIMARY KEY,
#     request_id TEXT,
#     model_version TEXT,
#     predicted_class INT,
#     confidence FLOAT,
#     probabilities JSONB,
#     timestamp TIMESTAMPTZ DEFAULT NOW()
# );

def log_prediction_to_db(conn, request_id, model_version, pred_class, confidence, probs):
    cur = conn.cursor()
    cur.execute(
        '''INSERT INTO prediction_log
           (request_id, model_version, predicted_class, confidence, probabilities)
           VALUES (%s, %s, %s, %s, %s)''',
        (request_id, model_version, pred_class, confidence, json.dumps(probs))
    )
    conn.commit()

print('DB logging function ready.')

import numpy as np

np.random.seed(42)
# Calibrated model: high confidence = high accuracy
y_true = np.random.binomial(1, 0.6, 10000)
base_prob = np.where(y_true == 1, np.random.beta(7, 2, 10000),
                                   np.random.beta(2, 7, 10000))
y_prob = np.clip(base_prob, 0, 1)

results = []
for threshold in np.arange(0.5, 0.96, 0.05):
    mask = y_prob > threshold
    if mask.sum() < 100:
        break
    acc = (y_true[mask] == (y_prob[mask] > 0.5)).mean()
    coverage = mask.mean()
    results.append({'threshold': round(threshold, 2), 'accuracy': round(acc, 3),
                    'coverage': round(coverage, 3)})

import pandas as pd
df = pd.DataFrame(results)
print(df.to_string(index=False))

import numpy as np

# Training period confidence stats (computed once during deployment)
train_conf_mean = 0.82
train_conf_std = 0.08
Z_THRESHOLD = 2.0  # alert if > 2 std below baseline

def check_confidence_alert(daily_confidences):
    daily_mean = np.mean(daily_confidences)
    z_score = (train_conf_mean - daily_mean) / train_conf_std

    status = 'ALERT' if z_score > Z_THRESHOLD else 'OK'
    print(f'Daily mean: {daily_mean:.4f}')
    print(f'Z-score below baseline: {z_score:.2f}')
    print(f'Status: {status}')
    return status

# Healthy day
check_confidence_alert(np.random.beta(8, 2, 500))  # mean ~0.8
print()
# Degraded day
check_confidence_alert(np.random.beta(3, 3, 500))  # mean ~0.5

import pandas as pd
import numpy as np

# Simulated prediction log
df = pd.DataFrame({
    'segment': np.random.choice(['new_user', 'returning', 'premium'], 1000, p=[0.3, 0.5, 0.2]),
    'confidence': np.random.beta(8, 2, 1000),
    'week': np.random.choice(range(1, 9), 1000)
})
# Simulate drop for new_user segment in weeks 6-8
mask = (df['segment'] == 'new_user') & (df['week'] >= 6)
df.loc[mask, 'confidence'] *= 0.6

print(df.groupby(['week', 'segment'])['confidence'].mean().round(3).to_string())