diff --git a/IMPLEMENTATION_PLAN.md b/IMPLEMENTATION_PLAN.md
index fcb3c64..db748c4 100644
--- a/IMPLEMENTATION_PLAN.md
+++ b/IMPLEMENTATION_PLAN.md
@@ -215,13 +215,17 @@ Comprehensive review and improvement of all Plotly charts in the Dash dashboard.
- **Checkpoint**: Dose distribution visible as box/violin plots
### D.3 Drug timeline (Gantt chart)
-- [ ] Create `get_drug_timeline()` query in `pathway_queries.py`:
+- [x] Create `get_drug_timeline()` query in `pathway_queries.py`:
- Level 3 nodes with `first_seen`, `last_seen`, `labels`, `value` per drug × directory
-- [ ] Create `create_drug_timeline_figure(data, title)` in plotly_generator.py:
+ - Aggregates across trusts: MIN(first_seen), MAX(last_seen), SUM(value), weighted avg cost_pp_pa
+ - Supports directory/trust filters
+- [x] Create `create_drug_timeline_figure(data, title)` in plotly_generator.py:
- Gantt-style using `go.Bar` (horizontal bars from first_seen to last_seen)
- - Grouped by directory, colored by patient count
-- [ ] Add "Timeline" tab to `TAB_DEFINITIONS` in `chart_card.py`
-- [ ] Add callback wiring
+ - One trace per bar, grouped by directory with legend grouping
+ - Colored by directory using `DRUG_PALETTE`, patient count as bar text
+ - Dynamic height (28px per bar), `_base_layout()` + `_smart_legend()`
+- [x] Add "Timeline" tab to `TAB_DEFINITIONS` in `chart_card.py` (8th tab)
+- [x] Add `_render_timeline()` helper + dispatch case in `chart.py`
- **Checkpoint**: Timeline tab shows when each drug cohort was active
### D.4 NICE TA compliance dashboard
diff --git a/dash_app/callbacks/chart.py b/dash_app/callbacks/chart.py
index eb6b2bf..f0c3eeb 100644
--- a/dash_app/callbacks/chart.py
+++ b/dash_app/callbacks/chart.py
@@ -367,6 +367,31 @@ def _render_network(app_state, title):
return create_drug_network_figure(data, title)
+def _render_timeline(app_state, title):
+ """Build the drug timeline Gantt chart from current filter state."""
+ from dash_app.data.queries import get_drug_timeline
+ from visualization.plotly_generator import create_drug_timeline_figure
+
+ filter_id = (app_state or {}).get("date_filter_id", "all_6mo")
+ chart_type = (app_state or {}).get("chart_type", "directory")
+
+ selected_dirs = (app_state or {}).get("selected_directorates") or []
+ selected_trusts = (app_state or {}).get("selected_trusts") or []
+ directory = selected_dirs[0] if len(selected_dirs) == 1 else None
+ trust = selected_trusts[0] if len(selected_trusts) == 1 else None
+
+ try:
+ data = get_drug_timeline(filter_id, chart_type, directory, trust)
+ except Exception:
+ log.exception("Failed to load drug timeline data")
+ return _empty_figure("Failed to load drug timeline data.")
+
+ if not data:
+ return _empty_figure("No drug timeline data available.\nTry adjusting your filters.")
+
+ return create_drug_timeline_figure(data, title)
+
+
def register_chart_callbacks(app):
"""Register tab switching, pathway data loading, and chart rendering callbacks."""
@@ -519,6 +544,9 @@ def register_chart_callbacks(app):
elif active_tab == "network":
fig = _render_network(app_state, title)
+ elif active_tab == "timeline":
+ fig = _render_timeline(app_state, title)
+
else:
# Placeholder for charts not yet implemented
tab_label = dict(TAB_DEFINITIONS).get(active_tab, active_tab)
diff --git a/dash_app/components/chart_card.py b/dash_app/components/chart_card.py
index 8e8c28a..24ef5a0 100644
--- a/dash_app/components/chart_card.py
+++ b/dash_app/components/chart_card.py
@@ -12,6 +12,7 @@ TAB_DEFINITIONS = [
("depth", "Depth"),
("scatter", "Scatter"),
("network", "Network"),
+ ("timeline", "Timeline"),
]
# Full set retained for Trust Comparison dashboard (Phase 10.8)
diff --git a/dash_app/data/queries.py b/dash_app/data/queries.py
index fc968d7..f2e829f 100644
--- a/dash_app/data/queries.py
+++ b/dash_app/data/queries.py
@@ -28,6 +28,7 @@ from data_processing.pathway_queries import (
get_pathway_depth_distribution as _get_pathway_depth_distribution,
get_duration_cost_scatter as _get_duration_cost_scatter,
get_drug_network as _get_drug_network,
+ get_drug_timeline as _get_drug_timeline,
)
DB_PATH = Path(__file__).resolve().parents[2] / "data" / "pathways.db"
@@ -227,3 +228,13 @@ def get_drug_network(
) -> dict:
"""Undirected drug co-occurrence network for network graph."""
return _get_drug_network(DB_PATH, date_filter_id, chart_type, directory, trust)
+
+
+def get_drug_timeline(
+ date_filter_id: str = "all_6mo",
+ chart_type: str = "directory",
+ directory: Optional[str] = None,
+ trust: Optional[str] = None,
+) -> list[dict]:
+ """Drug timeline data (first_seen, last_seen) for Gantt chart."""
+ return _get_drug_timeline(DB_PATH, date_filter_id, chart_type, directory, trust)
diff --git a/src/data_processing/pathway_queries.py b/src/data_processing/pathway_queries.py
index 6b16ae4..a47db71 100644
--- a/src/data_processing/pathway_queries.py
+++ b/src/data_processing/pathway_queries.py
@@ -1373,6 +1373,76 @@ def get_drug_network(
conn.close()
+def get_drug_timeline(
+ db_path: Path,
+ date_filter_id: str,
+ chart_type: str,
+ directory: Optional[str] = None,
+ trust: Optional[str] = None,
+) -> list[dict]:
+ """Get drug timeline data for Gantt-style chart.
+
+ Queries level 3 nodes and aggregates across trusts to get the earliest
+ first_seen and latest last_seen per drug × directory.
+
+ Returns list of dicts sorted by directory then first_seen:
+ [{"drug": "ADALIMUMAB", "directory": "RHEUMATOLOGY",
+ "first_seen": "2019-04-04T00:00:00", "last_seen": "2025-12-30T00:00:00",
+ "patients": 2053, "cost_pp_pa": 2170.5}, ...]
+ """
+ conn = sqlite3.connect(str(db_path))
+ conn.row_factory = sqlite3.Row
+ try:
+ conditions = ["date_filter_id = ?", "chart_type = ?", "level = 3"]
+ params: list = [date_filter_id, chart_type]
+
+ if directory:
+ conditions.append("directory = ?")
+ params.append(directory)
+ if trust:
+ conditions.append("trust_name = ?")
+ params.append(trust)
+
+ where = " AND ".join(conditions)
+ query = f"""
+ SELECT
+ labels AS drug,
+ directory,
+ MIN(first_seen) AS first_seen,
+ MAX(last_seen) AS last_seen,
+ SUM(value) AS patients,
+ CASE
+ WHEN SUM(value) > 0
+ THEN ROUND(SUM(CAST(cost_pp_pa AS REAL) * value) / SUM(value), 2)
+ ELSE 0
+ END AS cost_pp_pa
+ FROM pathway_nodes
+ WHERE {where}
+ AND first_seen IS NOT NULL AND first_seen != ''
+ AND last_seen IS NOT NULL AND last_seen != ''
+ GROUP BY labels, directory
+ ORDER BY directory, first_seen
+ """
+
+ rows = conn.execute(query, params).fetchall()
+ return [
+ {
+ "drug": r["drug"],
+ "directory": r["directory"],
+ "first_seen": r["first_seen"],
+ "last_seen": r["last_seen"],
+ "patients": r["patients"] or 0,
+ "cost_pp_pa": r["cost_pp_pa"] or 0,
+ }
+ for r in rows
+ if r["patients"] and r["patients"] > 0
+ ]
+ except sqlite3.Error:
+ return []
+ finally:
+ conn.close()
+
+
def get_directorate_summary(
db_path: Path,
date_filter_id: str,
diff --git a/src/visualization/plotly_generator.py b/src/visualization/plotly_generator.py
index 96d58ad..2943e4b 100644
--- a/src/visualization/plotly_generator.py
+++ b/src/visualization/plotly_generator.py
@@ -2085,3 +2085,124 @@ def create_drug_network_figure(data: dict, title: str = "") -> go.Figure:
fig.update_layout(**layout)
return fig
+
+
+def create_drug_timeline_figure(data: list[dict], title: str = "") -> go.Figure:
+ """Create a Gantt-style timeline showing when each drug cohort was active.
+
+ Each horizontal bar spans from first_seen to last_seen for a drug,
+ grouped by directory, with color indicating directory and text showing
+ patient count.
+
+ Args:
+ data: List of dicts with keys: drug, directory, first_seen, last_seen,
+ patients, cost_pp_pa.
+ title: Chart title.
+
+ Returns:
+ Plotly Figure with horizontal bars.
+ """
+ if not data:
+ return go.Figure()
+
+ from datetime import datetime
+
+ display_title = title or "Drug Timeline"
+
+ # Parse dates and sort by directory then first_seen
+ for d in data:
+ d["_fs"] = datetime.fromisoformat(d["first_seen"])
+ d["_ls"] = datetime.fromisoformat(d["last_seen"])
+ d["_duration_days"] = max(1, (d["_ls"] - d["_fs"]).days)
+
+ # Sort: by directory alphabetically, then by first_seen ascending
+ data.sort(key=lambda d: (d["directory"], d["_fs"]))
+
+ # Assign colors by directory
+ directories = list(dict.fromkeys(d["directory"] for d in data))
+ dir_colors = {
+ d: DRUG_PALETTE[i % len(DRUG_PALETTE)]
+ for i, d in enumerate(directories)
+ }
+
+ # Build y-axis labels: "Drug (Directory)" for multi-directory views, just "Drug" for single
+ single_directory = len(directories) == 1
+ y_labels = []
+ for d in data:
+ if single_directory:
+ y_labels.append(d["drug"])
+ else:
+ y_labels.append(f"{d['drug']} ({d['directory']})")
+
+ # Build one trace per directory for legend grouping
+ fig = go.Figure()
+ dir_legend_shown = set()
+
+ for i, d in enumerate(data):
+ show_legend = d["directory"] not in dir_legend_shown
+ dir_legend_shown.add(d["directory"])
+
+ duration_ms = d["_duration_days"] * 86_400_000 # days → milliseconds
+ patients = d["patients"]
+ cost = d["cost_pp_pa"]
+
+ fig.add_trace(
+ go.Bar(
+ y=[y_labels[i]],
+ x=[duration_ms],
+ base=[d["_fs"]],
+ orientation="h",
+ marker=dict(
+ color=dir_colors[d["directory"]],
+ line=dict(width=0),
+ ),
+ name=d["directory"],
+ legendgroup=d["directory"],
+ showlegend=show_legend,
+ text=f"{patients:,}",
+ textposition="inside",
+ textfont=dict(color="white", size=10),
+ hovertemplate=(
+ f"{d['drug']}
"
+ f"Directory: {d['directory']}
"
+ f"First seen: {d['_fs'].strftime('%b %Y')}
"
+ f"Last seen: {d['_ls'].strftime('%b %Y')}
"
+ f"Duration: {d['_duration_days']:,} days
"
+ f"Patients: {patients:,}
"
+ f"Cost p.a.: £{cost:,.0f}"
+ ""
+ ),
+ )
+ )
+
+ # Layout
+ n_bars = len(data)
+ bar_height = 28
+ dynamic_height = max(400, n_bars * bar_height + 120)
+
+ n_dirs = len(directories)
+ legend_margins = _smart_legend_margin(n_dirs)
+ legend = _smart_legend(n_dirs, legend_title="Directory")
+
+ layout = _base_layout(display_title)
+ layout.update(
+ xaxis=dict(
+ type="date",
+ gridcolor=GRID_COLOR,
+ dtick="M6",
+ tickformat="%b\n%Y",
+ ),
+ yaxis=dict(
+ automargin=True,
+ autorange="reversed",
+ tickfont=dict(size=11),
+ ),
+ barmode="overlay",
+ height=dynamic_height,
+ margin=dict(t=60, l=8, **legend_margins),
+ legend=legend,
+ bargap=0.3,
+ )
+ fig.update_layout(**layout)
+
+ return fig