Disaster Twitts: From Kaggle Notebook to First Professional ML Deployment

train_metrics = {"accuracy": BinaryAccuracy(threshold=0.5)}
val_metrics = {
    "accuracy": BinaryAccuracy(threshold=0.5),
    "precision": BinaryPrecision(threshold=0.5),
    "recall": BinaryRecall(threshold=0.5),
    "f1": BinaryF1Score(threshold=0.5),
}

criterion = nn.BCEWithLogitsLoss(pos_weight=pos_class_weight)
optimizer = optim.Adam(model.parameters(), lr=3e-4)

best_f1, best_t = find_best_threshold(model, val_dl)
torch.save(best_t, "best_threshold.pt")

def main():
    CFG = TrainConfig()
    seed_everything(CFG.SEED)

    df = pd.read_csv(CFG.TRAIN_CSV)
    df = preprocess_df(df, CFG)

    train_df, val_df = train_test_split(
        df,
        train_size=CFG.TRAIN_TEST_SPLIT,
        stratify=df["target"],
        random_state=CFG.SEED,
        shuffle=True,
    )

    vocab = build_vocab(train_df["final_text"], CFG)
    save_vocabs(vocab, CFG.VOCAB_PATH)

    train_loader = DataLoader(DisasterDataset(train_df, vocab, CFG), batch_size=128, shuffle=True)
    val_loader = DataLoader(DisasterDataset(val_df, vocab, CFG), batch_size=128, shuffle=False)

    embedding_matrix = load_embedding(CFG.GLOVE_PATH, vocab, CFG.EMB_DIM)
    model = DisasterTwittsClassifier(..., embedding=embedding_matrix).to(CFG.DEVICE)

@asynccontextmanager
async def lifespan(app: FastAPI):
    config = AppConfig()
    seed_everything(config.SEED)

    if not config.VOCAB_PATH.exists():
        raise ArtifactError("VOCAB_NOT_FOUND", "Vocabulary file not found", {"path": str(config.VOCAB_PATH)})
    if not config.MODEL_PATH.exists():
        raise ArtifactError("MODEL_NOT_FOUND", "Model file not found", {"path": str(config.MODEL_PATH)})

    word2idx, idx2word, vocab_size = load_vocabs(config.VOCAB_PATH)
    predictor = DisasterTwittsPredictor.from_config(config, vocab_size)
    predictor.load_label_mapping(config.LABEL_MAPPING_PATH)
    db_conn = connect_db(config.DB_PATH)
    init_db(db_conn)

    app.state.config = config
    app.state.word2idx = word2idx
    app.state.predictor = predictor
    app.state.db = db_conn
    yield
    db_conn.close()

def run_prediction(tweet: str, keyword: str | None = None) -> PredictResponse:
    final_text, warnings = validator.validate(tweet, keyword=keyword)
    input_ids, input_length = tokenize_and_pad(final_text, word2idx, config)
    prob, label, label_name = predictor.predict(input_ids, input_length, return_label_name=True)

    response = PredictResponse(
        probability=prob,
        label=label,
        label_name=label_name,
        threshold=predictor.threshold,
        warnings=list(warnings),
    )
    log_prediction(db_conn, tweet=tweet, keyword=keyword, final_text=final_text, ...)
    return response

{% set has_result = (result is not none) %}
{% set is_disaster = has_result and (result.label == 1) %}
{% set disaster_prob = ((result.probability * 100)|round(1)) if has_result else 0 %}

<form method="post" action="/predict-ui" class="card">
  <textarea id="tweet" name="tweet" rows="6" required>{{ form_data.tweet }}</textarea>
  <input id="keyword" name="keyword" type="text" value="{{ form_data.keyword }}" />
  <button type="submit">Predict</button>
</form>

{% if result.warnings %}
  <ul>{% for item in result.warnings %}<li>{{ item.warning_code }}: {{ item.message }}</li>{% endfor %}</ul>
{% endif %}

def log_prediction(conn, *, tweet, keyword, final_text, probability, label, label_name, threshold, warnings):
    warnings_json = json.dumps(warnings or [], ensure_ascii=True)
    created_at = datetime.now(timezone.utc).isoformat().replace("+00:00", "Z")
    conn.execute(
        "INSERT INTO prediction_logs (...) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)",
        (created_at, tweet, keyword, final_text, probability, label, label_name, threshold, warnings_json),
    )
    conn.commit()

FROM python:3.11-slim

ENV PYTHONDONTWRITEBYTECODE=1 \
    PYTHONUNBUFFERED=1 \
    PIP_NO_CACHE_DIR=1

WORKDIR /app

RUN addgroup --system appgroup \
    && adduser --system --ingroup appgroup appuser

COPY requirements.txt .
RUN python -m pip install --no-cache-dir --retries 5 --timeout 120 -r requirements.txt

COPY app ./app
COPY artifacts ./artifacts
COPY data ./data
RUN mkdir -p /app/data && chown -R appuser:appgroup /app

USER appuser
EXPOSE 8000
CMD ["uvicorn", "app.api:app", "--host", "0.0.0.0", "--port", "8000"]

services:
  app:
    build: .
    image: disaster_tweets:latest
    ports:
      - "${APP_PORT:-8000}:8000"
    environment:
      APP_ENV: ${APP_ENV:-production}
      DB_PATH: ${DB_PATH:-/app/data/predictions.db}
    volumes:
      - disaster_data:/app/data
    healthcheck:
      test: ["CMD", "python", "-c", "import urllib.request; urllib.request.urlopen('http://127.0.0.1:8000/docs', timeout=5)"]
      interval: 30s
      timeout: 10s
      retries: 3
      start_period: 20s
    restart: unless-stopped

volumes:
  disaster_data: