```python
#!/usr/bin/env python3
"""
Security Log Anomaly Detection Starter Script

Purpose:
    Build a simple anomaly detection baseline for semi-structured security logs.
    This script is intentionally vendor-neutral and safe by default.

What it does:
    - Reads CSV or JSON Lines security log data
    - Validates required columns
    - Aggregates events into entity/time windows
    - Computes practical features for anomaly detection
    - Fits a simple robust baseline using median and MAD-derived z-scores
    - Scores windows and flags the most unusual ones for analyst review
    - Exports results to CSV

What it does NOT do:
    - It does not connect to any SIEM, EDR, or cloud service
    - It does not send alerts automatically
    - It does not train a production-ready model
    - It does not replace threat hunting or rule-based detections

Recommended usage:
    python security_log_anomaly_starter.py \
        --input security_logs.csv \
        --output scored_windows.csv \
        --entity user \
        --time-column timestamp \
        --event-column event_type \
        --source-column src_ip

Input expectations:
    The input file should contain at least:
        - a timestamp column
        - an entity column (user, host, service_account, etc.)
        - optionally event type and source IP columns

Example CSV columns:
    timestamp,user,host,event_type,src_ip,action,success
    2026-01-01T09:00:00Z,alice,laptop-01,auth,10.0.0.5,login,1

Notes:
    - The script uses simple, explainable features by design.
    - It is suitable as a starting point for validation and workflow testing.
    - You should tune the window size and thresholds for your environment.
"""

from __future__ import annotations

import argparse
import json
import math
import sys
from dataclasses import dataclass
from pathlib import Path
from typing import Dict, Iterable, List, Optional, Sequence, Tuple

import pandas as pd


REQUIRED_MIN_COLUMNS = {"timestamp"}
DEFAULT_WINDOW = "4H"
DEFAULT_ALERT_PCT = 0.95
DEFAULT_OUTPUT = "scored_windows.csv"


@dataclass
class Config:
    input_path: Path
    output_path: Path
    entity_column: str
    time_column: str
    event_column: Optional[str]
    source_column: Optional[str]
    success_column: Optional[str]
    window: str
    alert_percentile: float
    top_n: int


def parse_args() -> Config:
    parser = argparse.ArgumentParser(
        description="Baseline and score anomalies in security log data.",
        formatter_class=argparse.ArgumentDefaultsHelpFormatter,
    )
    parser.add_argument("--input", required=True, help="Path to CSV or JSONL input file")
    parser.add_argument("--output", default=DEFAULT_OUTPUT, help="Path to output CSV")
    parser.add_argument("--entity", default="user", help="Entity column to group by")
    parser.add_argument("--time-column", default="timestamp", help="Timestamp column name")
    parser.add_argument("--event-column", default=None, help="Optional event type column")
    parser.add_argument("--source-column", default=None, help="Optional source IP or source identifier column")
    parser.add_argument("--success-column", default=None, help="Optional success/failure column")
    parser.add_argument("--window", default=DEFAULT_WINDOW, help="Time window for aggregation, e.g. 1H, 4H, 1D")
    parser.add_argument(
        "--alert-percentile",
        type=float,
        default=DEFAULT_ALERT_PCT,
        help="Percentile threshold for alerting on anomaly scores, between 0 and 1",
    )
    parser.add_argument("--top-n", type=int, default=20, help="Number of top anomalous windows to display")
    args = parser.parse_args()

    input_path = Path(args.input)
    output_path = Path(args.output)

    if not input_path.exists():
        parser.error(f"Input file does not exist: {input_path}")
    if args.entity.strip() == "":
        parser.error("--entity must not be empty")
    if args.time_column.strip() == "":
        parser.error("--time-column must not be empty")
    if not (0 < args.alert_percentile < 1):
        parser.error("--alert-percentile must be between 0 and 1")
    if args.top_n <= 0:
        parser.error("--top-n must be greater than 0")

    return Config(
        input_path=input_path,
        output_path=output_path,
        entity_column=args.entity,
        time_column=args.time_column,
        event_column=args.event_column,
        source_column=args.source_column,
        success_column=args.success_column,
        window=args.window,
        alert_percentile=args.alert_percentile,
        top_n=args.top_n,
    )


def load_input(path: Path) -> pd.DataFrame:
    suffix = path.suffix.lower()
    if suffix == ".csv":
        df = pd.read_csv(path)
    elif suffix in {".jsonl", ".ndjson"}:
        df = pd.read_json(path, lines=True)
    else:
        raise ValueError("Unsupported input format. Use CSV, JSONL, or NDJSON.")

    if df.empty:
        raise ValueError("Input file is empty.")
    return df


def validate_columns(df: pd.DataFrame, cfg: Config) -> None:
    required = {cfg.time_column, cfg.entity_column}
    missing = required - set(df.columns)
    if missing:
        raise ValueError(f"Missing required columns: {sorted(missing)}")

    if cfg.event_column and cfg.event_column not in df.columns:
        raise ValueError(f"Event column not found: {cfg.event_column}")
    if cfg.source_column and cfg.source_column not in df.columns:
        raise ValueError(f"Source column not found: {cfg.source_column}")
    if cfg.success_column and cfg.success_column not in df.columns:
        raise ValueError(f"Success column not found: {cfg.success_column}")


def prepare_frame(df: pd.DataFrame, cfg: Config) -> pd.DataFrame:
    frame = df.copy()
    frame[cfg.time_column] = pd.to_datetime(frame[cfg.time_column], errors="coerce", utc=True)
    frame = frame.dropna(subset=[cfg.time_column])

    if frame.empty:
        raise ValueError("No valid timestamps found after parsing.")

    frame[cfg.entity_column] = frame[cfg.entity_column].astype(str).fillna("unknown")

    if cfg.event_column:
        frame[cfg.event_column] = frame[cfg.event_column].astype(str).fillna("unknown")
    if cfg.source_column:
        frame[cfg.source_column] = frame[cfg.source_column].astype(str).fillna("unknown")
    if cfg.success_column:
        frame[cfg.success_column] = frame[cfg.success_column].astype(str).str.lower().fillna("unknown")

    return frame.sort_values(cfg.time_column)


def build_features(df: pd.DataFrame, cfg: Config) -> pd.DataFrame:
    frame = df.copy()
    frame = frame.set_index(cfg.time_column)

    group_cols = [cfg.entity_column]
    agg_dict: Dict[str, Tuple[str, str]] = {}

    agg_dict["event_count"] = (cfg.entity_column, "size")

    if cfg.event_column:
        agg_dict["unique_event_types"] = (cfg.event_column, pd.Series.nunique)
    else:
        frame["_event_placeholder"] = "event"
        agg_dict["unique_event_types"] = ("_event_placeholder", pd.Series.nunique)

    if cfg.source_column:
        agg_dict["unique_sources"] = (cfg.source_column, pd.Series.nunique)
    else:
        frame["_source_placeholder"] = "source"
        agg_dict["unique_sources"] = ("_source_placeholder", pd.Series.nunique)

    if cfg.success_column:
        success_series = frame[cfg.success_column].isin({"1", "true", "yes", "success"})
        frame["_fail_flag"] = (~success_series).astype(int)
        frame["_success_flag"] = success_series.astype(int)
        agg_dict["success_count"] = ("_success_flag", "sum")
        agg_dict["failure_count"] = ("_fail_flag", "sum")
    else:
        frame["_success_flag"] = 0
        frame["_fail_flag"] = 0
        agg_dict["success_count"] = ("_success_flag", "sum")
        agg_dict["failure_count"] = ("_fail_flag", "sum")

    grouped = frame.groupby([pd.Grouper(freq=cfg.window), cfg.entity_column]).agg(**agg_dict).reset_index()

    grouped = grouped.rename(columns={cfg.time_column: "window_start"})
    grouped["window_start"] = pd.to_datetime(grouped["window_start"], utc=True)

    grouped["failure_rate"] = grouped.apply(
        lambda r: r["failure_count"] / r["event_count"] if r["event_count"] > 0 else 0.0,
        axis=1,
    )
    grouped["source_diversity"] = grouped.apply(
        lambda r: r["unique_sources"] / r["event_count"] if r["event_count"] > 0 else 0.0,
        axis=1,
    )
    grouped["event_diversity"] = grouped.apply(
        lambda r: r["unique_event_types"] / r["event_count"] if r["event_count"] > 0 else 0.0,
        axis=1,
    )
    grouped["session_intensity"] = grouped["event_count"] / grouped["event_count"].replace(0, pd.NA).median()
    grouped["session_intensity"] = grouped["session_intensity"].fillna(0.0)

    return grouped


def robust_zscore(series: pd.Series) -> pd.Series:
    median = series.median()
    mad = (series - median).abs().median()
    if mad == 0 or pd.isna(mad):
        return (series - median).abs()
    return 0.6745 * (series - median).abs() / mad


def score_windows(features: pd.DataFrame) -> pd.DataFrame:
    scored = features.copy()
    numeric_cols = [
        "event_count",
        "unique_event_types",
        "unique_sources",
        "success_count",
        "failure_count",
        "failure_rate",
        "source_diversity",
        "event_diversity",
        "session_intensity",
    ]

    for col in numeric_cols:
        scored[f"z_{col}"] = robust_zscore(scored[col].astype(float))

    z_cols = [c for c in scored.columns if c.startswith("z_")]
    scored["anomaly_score"] = scored[z_cols].sum(axis=1)
    return scored.sort_values(["anomaly_score", "event_count"], ascending=[False, False])


def threshold_alerts(scored: pd.DataFrame, percentile: float) -> pd.Series:
    cutoff = scored["anomaly_score"].quantile(percentile)
    return scored["anomaly_score"] >= cutoff


def preview_top_rows(scored: pd.DataFrame, top_n: int) -> None:
    cols = [c for c in ["window_start", "user", "host", "event_count", "failure_count", "anomaly_score"] if c in scored.columns]
    print(scored[cols].head(top_n).to_string(index=False))


def save_output(scored: pd.DataFrame, output_path: Path) -> None:
    output_path.parent.mkdir(parents=True, exist_ok=True)
    scored.to_csv(output_path, index=False)


def main() -> int:
    try:
        cfg = parse_args()
        df = load_input(cfg.input_path)
        validate_columns(df, cfg)
        prepared = prepare_frame(df, cfg)
        features = build_features(prepared, cfg)
        scored = score_windows(features)
        scored["alert"] = threshold_alerts(scored, cfg.alert_percentile)
        save_output(scored, cfg.output_path)

        print(f"Saved scored windows to: {cfg.output_path}")
        print(f"Total windows scored: {len(scored)}")
        print(f"Alert threshold percentile: {cfg.alert_percentile}")
        print(f"Alert count: {int(scored['alert'].sum())}")
        print("\nTop suspicious windows:")
        preview_top_rows(scored[scored["alert"]], cfg.top_n)
        return 0
    except Exception as exc:
        print(f"Error: {exc}", file=sys.stderr)
        return 1


if __name__ == "__main__":
    raise SystemExit(main())
```

### Quick start

1. Save the file as `security-log-anomaly-starter.py`.
2. Install dependencies:

```bash
pip install pandas
```

3. Run it against a CSV or JSONL export of security logs.

### Suggested operational next steps

- Start with a narrow entity scope, such as user or host.
- Compare anomaly output to a small set of known benign maintenance windows.
- Review the top alerts with analysts before tuning thresholds.
- Track drift over time and retrain on a stable baseline.
- Pair the script with deterministic detections for known-bad behavior.

### Validation checklist

- Confirm timestamps parse consistently and in UTC.
- Verify the selected entity scope matches how analysts investigate incidents.
- Check whether maintenance, patching, and onboarding periods skew the baseline.
- Measure alert volume, not just model score distribution.
- Record analyst feedback for each reviewed alert.
- Revisit the window size and percentile threshold when the environment changes.

### Caution

This script is a starting point for experimentation and workflow validation. It is not a production detection system and should not be used as the sole control for security monitoring.