Vamshi Krishna Janagama Data Scientist

def mmr_search(self, query, k=5, fetch_k=20):
    """
    Maximal Marginal Relevance — balances relevance vs diversity.
    MMR(d) = λ·sim(d,q) - (1-λ)·max_{dj∈S} sim(d,dj)
    λ=0.7: 70% relevance weight, 30% diversity penalty
    """
    q_vec     = self._embed_query(query)
    scores,ids = self._index.search(q_vec, fetch_k)
    candidates = [(int(i), float(s)) for s,i in zip(scores[0],ids[0])]

    selected, sel_vecs = [], []
    for _ in range(min(k, len(candidates))):
        best_id, best_mmr = -1, -np.inf
        for doc_id, rel in candidates:
            if doc_id in [s[0] for s in selected]: continue
            red = max((np.dot(self._doc_vec(doc_id),v)
                       for v in sel_vecs), default=0.0)
            mmr = 0.7*rel - 0.3*red  # λ=0.7
            if mmr > best_mmr: best_mmr, best_id = mmr, doc_id
        selected.append((best_id, rel))
        sel_vecs.append(self._doc_vec(best_id))
    return [(self._docs[i], s) for i,s in selected]

class RAGEvaluator:
    """MRR, NDCG, Precision/Recall + LLM faithfulness judge"""

    def mrr(self, retrieved, relevant) -> float:
        rrs = []
        for ret, rel in zip(retrieved, relevant):
            rr = next((1/(i+1) for i,r in enumerate(ret)
                        if r in set(rel)), 0.0)
            rrs.append(rr)
        return float(np.mean(rrs))  # 0.83 vs 0.61 BM25

    def faithfulness_score(self, q, ctx, answer) -> float:
        # GPT-4 as judge: does answer contain ONLY context info?
        prompt = f"Score 0-10 faithfulness.\nQ:{q}\nCtx:{ctx}\nA:{answer}"
        raw  = self.llm([HumanMessage(content=prompt)]).content
        return json.loads(raw)["score"] / 10.0  # avg: 0.91

@app.post("/ingest")
async def ingest(req: IngestRequest):
    count = get_pipeline().ingest(req.texts, req.metadatas)
    return {"indexed": count}

@app.post("/query", response_model=QueryResponse)
async def query(req: QueryRequest):
    if req.stream:
        return StreamingResponse(
            get_pipeline().stream_query(req.question),
            media_type="text/event-stream"
        )
    result = get_pipeline().query(req.question)
    return QueryResponse(
        answer=result.answer,
        sources=[d.page_content[:120] for d in result.documents],
        latency_ms=result.latency_ms  # avg 142ms
    )

def _add_interaction_features(self, df):
    # Pairwise ratio + product for top-6 numeric features
    for i in range(len(nums)):
        for j in range(i+1, min(i+4, len(nums))):
            df = df.withColumn(
                f"{nums[i]}_div_{nums[j]}",
                F.col(nums[i]) / (F.col(nums[j]) + 1e-9)
            ).withColumn(
                f"{nums[i]}_mul_{nums[j]}",
                F.col(nums[i]) * F.col(nums[j])
            )  # +24 interaction features

def _add_window_features(self, df):
    # Rolling window on partition key (e.g. customer_id)
    w3 = Window.partitionBy("customer_id").rowsBetween(-3,-1)
    for col in self.cfg.numeric_cols[:4]:
        df = df.withColumn(f"{col}_lag1", F.lag(col,1).over(w3))
               .withColumn(f"{col}_roll3", F.avg(col).over(w3))
    # +10 window features → total 48

class DriftMonitor:
    """KS test + PSI drift detection on every batch"""

    def detect(self, df: DataFrame) -> dict:
        pdf = df.select(self.numeric_cols).toPandas()
        report = {}
        for col in self.numeric_cols:
            ref    = self._reference[col]
            actual = pdf[col].dropna().values

            ks_stat, ks_p = stats.ks_2samp(ref, actual)
            psi_val       = self._psi(ref, actual)
            drifted = (ks_p < 0.05) or (psi_val > 0.2)
            # PSI > 0.2 = significant drift → retrain trigger
            report[col] = {
                "ks_p": ks_p, "psi": psi_val,
                "drifted": drifted,
                "severity": "HIGH" if psi_val>0.25 else "MEDIUM"
            }
        return report

@app.websocket("/ws/stream")
async def stream_detection(ws: WebSocket):
    """Real-time scoring — client sends batches, server responds."""
    await ws.accept()
    try:
        while True:
            raw     = await ws.receive_text()
            payload = json.loads(raw)
            X       = np.array(payload["data"], dtype=np.float32)
            result  = get_engine().detect(X)
            await ws.send_json({
                "scores":     result.scores.tolist(),
                "anomalies":  result.labels.tolist(),
                "latency_ms": result.latency_ms  # ~12ms avg
            })
    except WebSocketDisconnect: pass

def build_features(df):
    for w in [30, 60, 90, 180]:
        df[f'logins_{w}d'] = df.groupby('customer_id')['login_event'] \
            .transform(lambda x: x.rolling(w).sum())
    df['open_tickets_60d'] = df.groupby('customer_id')['ticket_open'] \
        .transform(lambda x: x.rolling(60).sum())
    df['feature_adoption_pct'] = (
        df['features_used_90d'] / df['features_purchased']).clip(0,1)
    return df

def find_optimal_threshold(y_true, y_prob, beta=2.0):
    """F-beta optimisation -- beta=2 weights recall 2x precision."""
    precision, recall, thresholds = precision_recall_curve(y_true, y_prob)
    f_beta = (1 + beta**2) * precision * recall / (
              beta**2 * precision + recall + 1e-8)
    return thresholds[np.argmax(f_beta)]  # 0.38

def apply_id_mapping(df):
    # Bridge 3 patient ID formats:
    # - Scheduling: 7-digit MRN
    # - Ward Mgmt:  encounter-level ID
    # - EHR: composite key (MRN + DOB hash)
    return df.merge(PATIENT_XWALK,
        left_on=df.attrs['id_col'],
        right_on=df.attrs['id_type'],
        how='left')

class ReadmissionEnsemble:
    def predict_proba(self, X):
        p_xgb  = self.xgb.predict_proba(X)[:,1]
        p_lgbm = self.lgbm.predict_proba(X)[:,1]
        return 0.55 * p_xgb + 0.45 * p_lgbm

def stratify_risk(scores):
    return pd.cut(scores, bins=[0, 0.35, 0.65, 1.0],
        labels=['LOW', 'MEDIUM', 'HIGH'])
    # HIGH  -> care coordinator + 48-hr post-discharge call
    # MEDIUM -> 7-day follow-up + pharmacy counselling
    # LOW   -> standard discharge pathway