Cognitive Talent Matrix (CTM)

Intelligent candidate ranking API. POST a Job Description, get back a ranked .xlsx of candidates sorted by a fused score combining:

1. Semantic similarity (PyTorch + sentence-transformers all-MiniLM-L6-v2) — 60% default weight
2. Deterministic metadata score (graduated experience curve) — 25% default weight
3. Behavioral signal (graduated project-throughput curve) — 15% default weight

final_score = w_semantic * semantic_score
            + w_metadata * metadata_score
            + w_behavioral * behavioral_score

Weights are now tunable per-request (Iter 5). All three must sum to ~1.0 (validated by Pydantic).

---

Quickstart

# 1. Install dependencies (CPU-only torch is fine for this workload)
pip install -r requirements.txt

# 2. Run the server
uvicorn main:app --reload --port 8000

# 3. Open the interactive docs
#    http://localhost:8000/docs

A sample_candidates.csv is included so the API works out of the box with no additional setup.

---

Endpoints

Method	Path	Purpose
GET	/health	Liveness probe + config introspection (weights, thresholds)
GET	/metrics	Prometheus text exposition (cache hits/misses, pipeline time)
POST	/rank-candidates/	JSON body → ranked .xlsx
POST	/rank-candidates/upload	Multipart form (jd_text + file) → ranked .xlsx

Example: JSON endpoint

curl -X POST http://localhost:8000/rank-candidates/ \
  -H "Content-Type: application/json" \
  -d '{
    "jd_text": "Senior Python backend engineer with FastAPI, PostgreSQL, Kubernetes experience",
    "candidates_file_path": "sample_candidates.csv"
  }' \
  --output ranked.xlsx

Example: Upload endpoint

curl -X POST http://localhost:8000/rank-candidates/upload \
  -F "jd_text=Senior Python backend engineer with FastAPI experience" \
  -F "file=@sample_candidates.csv" \
  --output ranked.xlsx

Example: Custom fusion weights (Iter 5)

curl -X POST http://localhost:8000/rank-candidates/ \
  -H "Content-Type: application/json" \
  -d '{
    "jd_text": "Senior Python backend engineer",
    "candidates_file_path": "sample_candidates.csv",
    "weights": {"semantic": 0.5, "metadata": 0.3, "behavioral": 0.2}
  }' \
  --output ranked.xlsx

---

Scoring math

Graduated curves (Iter 1 — replaces binary thresholds)

metadata_score   = 0.5 + 0.5 * min(years / 3, 1.0)       # [0.5, 1.0]
behavioral_score = 0.5 + 0.5 * min(projects / 3, 1.0)    # [0.5, 1.0]

Sample points:

years	metadata_score	projects	behavioral_score
0	0.500	0	0.500
1	0.667	1	0.667
2	0.833	2	0.833
3+	1.000	3+	1.000

Cosine similarity (PyTorch)

cos(jd, candidate) = (jd_norm · candidate_norm)
                  = (jd / ||jd||) · (candidate / ||candidate||)

Negative cosines are clamped to 0 (no actionable signal in "opposite meaning").

Deterministic tie-breaking (Iter 4)

Sort key (all descending): (final_score, semantic_score, years_of_experience) with kind="stable" so candidates tied on all three keys retain ingestion order. Output is fully reproducible across runs.

---

Architecture

POST /rank-candidates/
        │
        ▼
[async endpoint] ── asyncio.to_thread ──► [_rank_pipeline_sync]
                                              │
                                              ▼
                                   ┌─────────────────────┐
                                   │ 1. load_candidates  │  CSV / JSON / inline list
                                   ├─────────────────────┤
                                   │ 2. preprocess       │  NaN-fill + master_profile
                                   ├─────────────────────┤
                                   │ 3. embeddings       │  JD cached (Iter 3)
                                   │    • get_jd_embedding            │
                                   │    • generate_embeddings         │
                                   ├─────────────────────┤
                                   │ 4. compute scores   │  semantic / metadata / behavioral
                                   ├─────────────────────┤
                                   │ 5. compute_final    │  linear fusion (Iter 5 weights)
                                   ├─────────────────────┤
                                   │ 6. sort             │  stable, multi-key (Iter 4)
                                   ├─────────────────────┤
                                   │ 7. build export df  │  + rank column
                                   ├─────────────────────┤
                                   │ 8. write .xlsx      │  in-memory BytesIO
                                   └─────────────────────┘
                                              │
                                              ▼
                                   StreamingResponse (.xlsx)

Iter 6: the entire pipeline is wrapped in RANK_PIPELINE_SECONDS.time() so /metrics exposes p50/p95/p99 latency.

---

Observability (Iter 6)

GET /metrics returns Prometheus text format with:

Metric	Type	Description
jd_cache_hits_total	Counter	JD embedding cache hits
jd_cache_misses_total	Counter	JD embedding cache misses (encoder invoked)
rank_pipeline_seconds	Histogram	Full pipeline wall-clock time (load → export)

Example Prometheus scrape config:

scrape_configs:
  - job_name: 'ctm'
    static_configs:
      - targets: ['localhost:8000']

---

Testing (Iter 7)

pip install -r requirements.txt   # includes pytest + httpx
pytest tests/ -v

Test suite (55 tests, ~16s on CPU):

File	Tests	Scope
tests/test_scoring.py	13	Pure-numpy scoring math (graduated curves + fusion)
tests/test_preprocessing.py	8	master_profile construction, NaN handling, coercion
tests/test_weights_validation.py	12	FusionWeights pydantic validation (sum to 1.0, range)
tests/test_api.py	22	End-to-end: /health, /metrics, /rank-candidates/*

Key tests worth highlighting:

• test_top_candidate_is_alice_for_python_jd — canonical "does it actually work?" test
• test_ranking_is_deterministic — same input twice → identical output (Iter 4)
• test_jd_cache_miss_then_hit — asserts cache behavior via /metrics counters (Iter 3 + 6)
• test_custom_weights_change_ranking — proves per-request weights reorder output (Iter 5)

---

Project layout

.
├── main.py                    # FastAPI app — the complete backend
├── requirements.txt           # Pinned dependencies
├── sample_candidates.csv      # 12-candidate test dataset
├── README.md                  # This file
└── tests/
    ├── __init__.py
    ├── conftest.py            # Shared fixtures (TestClient, sample data, cache reset)
    ├── test_scoring.py        # Graduated curves + fusion math
    ├── test_preprocessing.py  # Data pipeline
    ├── test_weights_validation.py  # FusionWeights pydantic model
    └── test_api.py            # End-to-end HTTP tests

---

Iteration history

Iter	Feature
1	Graduated heuristic curves (replaces binary 0.5/1.0 thresholds)
2	POST /rank-candidates/upload multipart endpoint
3	JD embedding cache (SHA256-keyed, thread-safe, FIFO eviction)
4	Deterministic tie-breaker (final_score → semantic → years)
5	Per-request fusion weights via FusionWeights pydantic model
6	Prometheus metrics (/metrics endpoint, 3 custom metrics)
7	Pytest + TestClient suite (55 tests across 4 files)