"""A new, fold-local evaluation. No hidden test outcomes or leaderboard feedback.

Run with Python, pandas, numpy, scipy, scikit-learn and xgboost.
Pass --data-dir, --output-dir and --history for portable execution. This script never submits.
"""
from pathlib import Path
import hashlib
import json
import platform
import warnings
import numpy as np
import pandas as pd
from scipy.stats import pearsonr
from sklearn.ensemble import RandomForestClassifier, ExtraTreesClassifier
from sklearn.linear_model import LogisticRegression
from sklearn.pipeline import make_pipeline
from sklearn.preprocessing import StandardScaler
from sklearn.model_selection import StratifiedKFold, StratifiedGroupKFold, train_test_split
from sklearn.metrics import accuracy_score, brier_score_loss, confusion_matrix
from xgboost import XGBClassifier
import sklearn
import xgboost

ROOT=Path(globals().get('__file__', 'contender.py')).resolve().parent
warnings.filterwarnings('ignore', category=FutureWarning)

def title(frame):
    return frame.Name.str.extract(r' ([A-Za-z]+)\.',expand=False).replace({'Mlle':'Miss','Ms':'Miss','Mme':'Mrs'}).map(lambda x:x if x in ['Mr','Mrs','Miss','Master'] else 'Rare')

def features(reference, target, family=True):
    """Fit imputations and label-derived features exclusively on reference."""
    out=target[['Pclass','Age','Fare','SibSp','Parch']].copy()
    titles=title(target)
    median_age=reference.assign(Title=title(reference)).groupby('Title').Age.median()
    out['Age']=out.Age.fillna(titles.map(median_age)).fillna(reference.Age.median())
    out['Fare']=out.Fare.fillna(reference.Fare.median())
    out['Sex_Enc']=(target.Sex=='male').astype(int)
    out['FamilySize']=target.SibSp+target.Parch+1
    out['IsAlone']=(out.FamilySize==1).astype(int)
    out['Embarked_Enc']=target.Embarked.fillna('S').map({'S':0,'C':1,'Q':2})
    out['Title_Enc']=titles.map({'Mr':0,'Miss':1,'Mrs':2,'Master':3,'Rare':4})
    if family:
        ref_surnames=reference.Name.str.split(',').str[0]
        ref_fares=reference.Fare.fillna(reference.Fare.median())
        rates=[]
        for row in target.itertuples():
            fare=reference.Fare.median() if pd.isna(row.Fare) else row.Fare
            mask=(ref_surnames==row.Name.split(',')[0]) & (reference.PassengerId!=row.PassengerId) & ((ref_fares-fare).abs()<5)
            rates.append(float(reference.loc[mask,'Survived'].mean()) if mask.any() else .5)
        out['FamilySurvived']=rates
    return out.astype(float)

def models(seed=42):
    return {
        'Logistic regression':make_pipeline(StandardScaler(),LogisticRegression(C=1,max_iter=2000,random_state=seed)),
        'Random forest':RandomForestClassifier(n_estimators=300,max_depth=5,min_samples_leaf=5,random_state=seed,n_jobs=2),
        'XGBoost':XGBClassifier(n_estimators=220,max_depth=3,learning_rate=.04,subsample=.8,colsample_bytree=.8,reg_lambda=3,random_state=seed,n_jobs=2,eval_metric='logloss'),
        'Extra trees':ExtraTreesClassifier(n_estimators=300,max_depth=7,min_samples_leaf=3,random_state=seed,n_jobs=2),
    }

def predict_all(train, target, family=True, leaked_reference=None):
    ref=train if leaked_reference is None else leaked_reference
    x=features(ref,train,family); xt=features(ref,target,family)
    result={name:model.fit(x,train.Survived).predict_proba(xt)[:,1] for name,model in models().items()}
    result['Simple ensemble']=np.mean([result[n] for n in ['Logistic regression','Random forest','XGBoost']],axis=0)
    result['Sex-only baseline']=(target.Sex=='female').to_numpy(dtype=float)
    return result

def evaluate_cv(data, splits, family=True, leak=False):
    results={}
    for fold,(tr,va) in enumerate(splits):
        for name,prob in predict_all(data.iloc[tr],data.iloc[va],family,data if leak else None).items():
            result=results.setdefault(name,{'correct':0,'n':0,'folds':[],'brierSum':0})
            predictions=prob>.5
            correct=int((predictions==data.iloc[va].Survived).sum())
            result['correct']+=correct; result['n']+=len(va)
            result['folds'].append(correct/len(va))
            result['brierSum']+=float(np.square(prob-data.iloc[va].Survived).sum())
    return {name:{'accuracy':r['correct']/r['n'],'foldScores':r['folds'],'correct':r['correct'],'n':r['n'],'brier':r['brierSum']/r['n']} for name,r in results.items()}

def group_ids(data):
    """Connected ticket/surname groups prevent related passengers crossing folds."""
    parent=list(range(len(data)))
    def find(i):
        while parent[i]!=i: parent[i]=parent[parent[i]]; i=parent[i]
        return i
    for series in [data.Ticket.fillna(''),data.Name.str.split(',').str[0]]:
        seen={}
        for i,key in enumerate(series):
            if not key: continue
            if key in seen: parent[find(i)]=find(seen[key])
            else: seen[key]=i
    return [find(i) for i in range(len(data))]

def main(data_dir=ROOT, output_dir=ROOT, history=None):
    data_dir=Path(data_dir); output_dir=Path(output_dir)
    output_dir.mkdir(parents=True, exist_ok=True)
    train=pd.read_csv(data_dir/'train.csv'); test=pd.read_csv(data_dir/'test.csv')
    assert len(train)==891 and len(test)==418 and train.Survived.sum()==342
    assert 'Survived' not in test and not set(train.PassengerId)&set(test.PassengerId)
    assert test.PassengerId.tolist()==list(range(892,1310))
    assert train.isna().sum()[['Age','Cabin','Embarked']].tolist()==[177,687,2]
    assert test.isna().sum()[['Age','Cabin','Fare']].tolist()==[86,327,1]
    # Reserve labels before comparing any candidate. Never use them for selection.
    dev_idx,hold_idx=train_test_split(np.arange(len(train)),test_size=.2,stratify=train.Survived,random_state=804)
    dev=train.iloc[dev_idx].reset_index(drop=True)
    hold=train.iloc[hold_idx].reset_index(drop=True)
    splits=list(StratifiedKFold(5,shuffle=True,random_state=42).split(dev,dev.Survived))
    development=evaluate_cv(dev,splits)
    candidates=[n for n in development if n!='Sex-only baseline']
    winner=max(candidates,key=lambda n:(development[n]['accuracy'],-development[n]['brier']))
    print('Frozen selection:',winner,development[winner]['accuracy'],flush=True)
    hold_probs=predict_all(dev,hold)[winner]
    hold_pred=hold_probs>.5
    holdout={'n':len(hold),'correct':int((hold_pred==hold.Survived).sum()),'accuracy':float(accuracy_score(hold.Survived,hold_pred)),'brier':float(brier_score_loss(hold.Survived,hold_probs)),'confusionMatrix':confusion_matrix(hold.Survived,hold_pred).tolist(),'baselineAccuracy':float(accuracy_score(hold.Survived,hold.Sex=='female'))}
    # Separate diagnostics, performed after freezing winner and holdout result.
    full_splits=list(StratifiedKFold(5,shuffle=True,random_state=42).split(train,train.Survived))
    safe=evaluate_cv(train,full_splits)
    leaky=evaluate_cv(train,full_splits,leak=True)
    without_family=evaluate_cv(train,full_splits,family=False)
    groups=group_ids(train)
    group_splits=list(StratifiedGroupKFold(5,shuffle=True,random_state=42).split(train,train.Survived,groups))
    grouped=evaluate_cv(train,group_splits)
    final_probs=predict_all(train,test)[winner]
    final=(final_probs>.5).astype(int)
    out=pd.DataFrame({'PassengerId':test.PassengerId,'Survived':final})
    public=output_dir
    out.to_csv(public/'contender-submission.csv',index=False)
    if history is None:
        history=json.loads((ROOT/'history.json').read_text())['attempts']
    pairs=[(a['survivors'],a['score']) for a in history]
    r,p=pearsonr(*zip(*pairs)); rx,px=pearsonr(*zip(*pairs[1:]))
    # Illustration under independent Bernoulli assumptions, not a model ceiling.
    phat=339/418; z=1.96; denom=1+z*z/418
    centre=(phat+z*z/(2*418))/denom
    half=z*np.sqrt(phat*(1-phat)/418+z*z/(4*418**2))/denom
    stats={'all23Correlation':float(r),'all23PearsonP':float(p),'excludingFirstCorrelation':float(rx),'excludingFirstPearsonP':float(px),'wilson95':[centre-half,centre+half],'onePredictionPercentagePoints':100/418,'currentCorrect':339,'currentErrors':79}
    report={'runDate':'2026-09-05','method':'Five prespecified model candidates; development-only five-fold selection; locked 20% holdout (seed 804); separate full-data leakage, ablation and grouped diagnostics. Fixed >0.5 threshold. No leaderboard queries or hidden outcomes used. Holdout rows are freshly withheld for this run, not historically untouched by the original notebook.','development':development,'winner':winner,'holdout':holdout,'safeCV':safe,'leakyCV':leaky,'withoutFamilyCV':without_family,'groupedCV':grouped,'stats':stats,'submission':{'file':'contender-submission.csv','rows':418,'survivors':int(final.sum()),'publicScore':None,'status':'Prepared locally; not submitted to Kaggle'},'versions':{'python':platform.python_version(),'sklearn':sklearn.__version__,'xgboost':xgboost.__version__,'numpy':np.__version__,'pandas':pd.__version__},'data':{'source':'https://github.com/agconti/kaggle-titanic/tree/master/data','boundary':'Public mirror of competition train/test files; official direct download requires authentication. Dimensions, ID ranges, training outcome total and missing-value counts verified; official byte parity unverified.','trainSHA256':hashlib.sha256((data_dir/'train.csv').read_bytes()).hexdigest(),'testSHA256':hashlib.sha256((data_dir/'test.csv').read_bytes()).hexdigest(),'missingTrain':train.isna().sum().to_dict(),'missingTest':test.isna().sum().to_dict()},'limitations':['One holdout split; no claim of a statistical maximum or superiority on Kaggle.','Family/ticket grouping is a conservative proxy and can group unrelated surnames.','Leakage experiment isolates reference-label contamination in a new controlled implementation; it does not reproduce the original dependency environment.','Repeated diagnostic folds overlap; fold scores are not independent replications.']}
    (public/'evaluation.json').write_text(json.dumps(report,indent=2)+'\n')
    print(json.dumps({'winner':winner,'holdout':holdout,'stats':stats,'submission':report['submission']},indent=2),flush=True)

if __name__=='__main__':
    import argparse
    parser=argparse.ArgumentParser(description=__doc__)
    parser.add_argument('--data-dir',type=Path,default=ROOT)
    parser.add_argument('--output-dir',type=Path,default=ROOT)
    parser.add_argument('--history',type=Path,default=ROOT/'history.json')
    args=parser.parse_args()
    main(args.data_dir,args.output_dir,json.loads(args.history.read_text())['attempts'])