Files
Stride/app/analysis_engine.py

733 lines
28 KiB
Python
Raw Permalink Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""Owner-scoped analysis over compact run records; no raw samples are loaded."""
from collections import defaultdict, deque
from datetime import UTC, datetime, timedelta
from itertools import pairwise
from types import SimpleNamespace
from zoneinfo import ZoneInfo
from .extensions import db
from .models import Activity, DailyHealth, User
from .stats import METRICS, format_value, metric_value, period_bounds
from .stats import aggregate as legacy_aggregate
from .stats_analysis import describe, evaluate, expression_tree, statistic, valid
from .stats_dsl import QueryError
def catalog_for(q):
catalog = dict(METRICS)
for spec in q["derived"]:
catalog[spec["name"]] = (
spec["expression"] if spec.get("inline") else spec["name"].replace("_", " ").capitalize(),
spec["unit"],
f"Per-run calculation: {spec['expression']}. Missing inputs and zero divisors produce missing data.",
)
return catalog
def value(row, key):
return row.values.get(key)
def summarize(rows, key, q):
spec = q["aggregates"].get(key)
if spec:
function, field = spec["function"], spec["field"]
if function == "count" and not field:
return len(rows), len(rows)
values = [value(row, field) for row in rows]
coverage = sum(valid(number) for number in values)
if function == "latest":
last = max(rows, key=lambda r: (r.instant, r.id)) if rows else None
return (value(last, field) if last else None), coverage
if function == "weighted_mean":
pairs = [
(value(row, field), row.distance_m if field == "pace" else row.duration_s)
for row in rows
if valid(value(row, field))
]
weight = sum(w for _, w in pairs)
return (sum(v * w for v, w in pairs) / weight if weight > 0 else None), coverage
return statistic(values, function), coverage
if key in {"count", "calories"}:
numbers = [value(row, key) for row in rows if valid(value(row, key))]
return (sum(numbers) if numbers or key == "count" else None), len(numbers)
if key in {
"pace",
"heart_rate",
"distance",
"duration",
"elevation",
"cadence",
"power",
"ground_contact",
"stride_length",
"vertical_oscillation",
}:
return legacy_aggregate(rows, key)
numbers = [value(row, key) for row in rows if valid(value(row, key))]
return statistic(numbers, "mean"), len(numbers)
def formatted(number, key, q):
if not valid(number):
return "—"
spec = q["aggregates"].get(key)
raw = spec["field"] if spec and spec["function"] not in {"count", "stddev", "sum"} else key
if raw in {"pace", "gap", "effort_time"} or any(s["name"] == raw and s["unit"] == "time" for s in q["derived"]):
return format_value(number, "pace")
if key == "count" or (spec and spec["function"] == "count"):
return f"{number:,.0f}"
return (
f"{number:,.2f}"
if key in {"stride_length", "gap_since_previous_run"}
or key in {s["name"] for s in q["derived"]}
else f"{number:,.1f}"
)
def condition_matches(number, condition):
if not valid(number):
return False
threshold = condition["value"]
return {
">": number > threshold,
">=": number >= threshold,
"<": number < threshold,
"<=": number <= threshold,
"=": number == threshold,
"!=": number != threshold,
}[condition["op"]]
def matches(row, conditions):
for condition in conditions:
op, field = condition["op"], condition["field"]
if op == "contains":
if condition["value"].casefold() not in row.title.casefold():
return False
elif op == "has":
if not valid(value(row, field)):
return False
elif op == "missing":
if valid(value(row, field)):
return False
elif not condition_matches(value(row, field), condition):
return False
return True
def ordered(rows, ordering):
key, direction = ordering
if key == "date":
return sorted(rows, key=lambda row: (row.instant, row.id), reverse=direction == "desc")
available = [row for row in rows if valid(value(row, key))]
absent = [row for row in rows if not valid(value(row, key))]
return sorted(
available,
key=lambda row: (value(row, key), row.instant, row.id),
reverse=direction == "desc",
) + ordered(absent, ["date", "desc"])
def prepare(raw, zone, q, catalog):
rows = []
previous, last_day, streak = None, None, 0
trees = {
spec["name"]: expression_tree(spec["expression"], catalog)[0]
for spec in q["derived"]
if not any(key in spec for key in ("window", "since", "best_distance"))
}
custom_windows = {spec["name"]: deque() for spec in q["derived"] if "window" in spec}
last_matching = {}
for record in raw:
row = SimpleNamespace(**record._mapping)
row.instant = (
row.started_at.replace(tzinfo=UTC)
if row.started_at.tzinfo is None
else row.started_at.astimezone(UTC)
)
row.local = row.instant.astimezone(zone)
row.day = row.local.date()
row.values = {key: metric_value(row, key) for key in METRICS}
row.values.update(
count=1,
speed=row.distance_m / row.duration_s * 3.6 if row.duration_s > 0 else None,
elevation_per_km=row.elevation_m / (row.distance_m / 1000)
if row.distance_m > 0
else None,
max_heart_rate=(row.analysis_summary or {}).get("max_hr")
or (row.analysis_summary or {}).get("summary", {}).get("max_hr")
or (row.metrics or {}).get("max_heart_rate"),
calories=(row.metrics or {}).get("total_calories"),
training_effect=(row.metrics or {}).get("total_training_effect"),
)
row.values.update({key: (row.session_context or {}).get(key)
for key in ("perceived_effort", "fatigue")})
for field in (
"pace",
"heart_rate",
"max_heart_rate",
"cadence",
"power",
"ground_contact",
"stride_length",
"vertical_oscillation",
):
if not valid(row.values[field]) or row.values[field] <= 0:
row.values[field] = None
row.previous = previous
row.values["gap_since_previous_run"] = (
(row.instant - previous.instant).total_seconds() / 86400 if previous else None
)
for key, field in (
("distance_change", "distance"),
("pace_change", "pace"),
("heart_rate_change", "heart_rate"),
):
now, before = row.values[field], previous.values[field] if previous else None
row.values[key] = now - before if valid(now) and valid(before) else None
if row.day != last_day:
streak = streak + 1 if last_day and (row.day - last_day).days == 1 else 1
last_day = row.day
row.values["run_day_streak"] = streak
for spec in q["derived"]:
name = spec["name"]
if "best_distance" in spec:
efforts = [e.get("seconds") for e in row.best_efforts or []
if e.get("distance") == spec["best_distance"] and valid(e.get("seconds"))]
row.values[name] = min(efforts) if efforts else None
continue
if "since" in spec:
if matches(row, spec["since"]):
last_matching[name] = row.day
last = last_matching.get(name)
row.values[name] = (row.day - last).days if last else None
continue
if "window" not in spec:
row.values[name] = evaluate(trees[name], row.values)
continue
window = custom_windows[name]
config = spec["window"]
while window and (row.instant - window[0][0]).total_seconds() >= config["days"] * 86400:
window.popleft()
window.append((row.instant, row.values[config["field"]]))
row.values[name] = statistic([item[1] for item in window], config["function"])
rows.append(row)
previous = row
return rows
def explore(owner_id, timezone, q, page=1, selected_bucket=None, export=False, raw_group=False):
from .training_sources import evidence as evidence_coverage
zone, catalog = ZoneInfo(timezone), catalog_for(q)
start, end, prior_start = period_bounds(q, timezone)
prior_end = start
if q["compare"] and q["comparison"] == "year":
import calendar
def previous_year(day):
return day.replace(
year=day.year - 1, day=min(day.day, calendar.monthrange(day.year - 1, day.month)[1])
)
prior_start, prior_end = previous_year(start), previous_year(end)
boundary = lambda day: datetime.combine(day, datetime.min.time(), zone).astimezone(UTC)
# Temporal columns use every earlier run, independently of query filters.
statement = (
db.select(
Activity.id,
Activity.title,
Activity.started_at,
Activity.distance_m,
Activity.duration_s,
Activity.elevation_m,
Activity.avg_hr,
Activity.analysis_summary,
Activity.metrics,
Activity.best_efforts,
Activity.session_context,
*([Activity.route_geometry] if q["route"] else []),
)
.where(
Activity.owner_id == owner_id,
Activity.sport == "running",
Activity.started_at < boundary(end),
)
.order_by(Activity.started_at, Activity.id)
)
if q["source"] not in {"runs", "days", "health"} and not q["all_time"]:
lower = min(
start,
prior_start if q["compare"] else start,
start - timedelta(days=q["baseline"] or 0),
)
statement = statement.where(Activity.started_at >= boundary(lower))
raw = db.session.execute(statement).all()
history = prepare(raw, zone, q, catalog)
user = db.session.get(User, owner_id)
if user and user.hr_bands:
for row in history:
intervals = (row.analysis_summary or {}).get("intervals", [])
usable = any(unit.get("hr_seconds") for unit in intervals)
for name in ("lower", "middle", "upper"):
row.values[name + "_hr_minutes"] = 0.0 if usable else None
for unit in intervals:
for hr, seconds in unit.get("hr_seconds", {}).items():
hr = float(hr)
name = "lower" if hr < user.hr_bands[0] else "middle" if hr < user.hr_bands[1] else "upper"
row.values[name + "_hr_minutes"] += seconds / 60
if q["route"]:
from .route_benchmarks import similar
reference = db.session.scalar(db.select(Activity.route_geometry).where(
Activity.id == q["route"], Activity.owner_id == owner_id))
if reference is None:
raise QueryError("Choose one of your own recorded routes.")
history = [row for row in history if similar(reference, row.route_geometry)]
if q["all_time"] and history:
start = history[0].day
prior_start = start
if q["all_time"] and q["source"] in {"days", "health"}:
first_health = db.session.scalar(db.select(db.func.min(DailyHealth.day)).where(
DailyHealth.owner_id == owner_id, DailyHealth.day < end))
if first_health:
start = min(start, first_health) if history else first_health
prior_start = start
if q["source"] in {"days", "health"}:
from .training_sources import daily
lower = min(start, prior_start if q["compare"] else start,
start - timedelta(days=q["baseline"] or 0))
lower -= timedelta(days=max((spec.get("window", {}).get("days", 0)
for spec in q["derived"]), default=0))
history = daily(history, owner_id, zone, lower, end, q, catalog)
elif q["source"] != "runs":
from .stats_sources import expand
history = expand(history, q, catalog)
def included(row):
return (
row.distance_m >= q["minimum"] * 1000
and (q["maximum"] is None or row.distance_m <= q["maximum"] * 1000)
and (q["hr_min"] is None or valid(row.avg_hr) and row.avg_hr >= q["hr_min"])
and (q["hr_max"] is None or valid(row.avg_hr) and row.avg_hr <= q["hr_max"])
and matches(row, q["filters"])
and (not q["filter_groups"] or any(matches(row, group) for group in q["filter_groups"]))
)
if q["reduce"]:
from .training_sources import reduce_runs
history = reduce_runs([row for row in history if included(row)], q["reduce"], catalog)
included = lambda row: True
current = ordered(
[row for row in history if start <= row.day < end and included(row)], q["sort"]
)
prior = ordered(
[
row
for row in history
if q["compare"] and prior_start <= row.day < prior_end and included(row)
],
q["sort"],
)
if q["take"]:
current, prior = current[: q["take"]], prior[: q["take"]]
keys = [
q["metric"],
*([] if q["secondary"] == "none" else [q["secondary"]]),
*q["extra_metrics"],
]
keys = list(dict.fromkeys(keys))
group, width = q["group"], q["bin"]
def bucket(row, offset=timedelta()):
day = row.day + offset
if group == "day":
return day
if group == "week":
return day - timedelta(days=day.weekday())
if group == "month":
return day.replace(day=1)
if group == "year":
return day.replace(month=1, day=1)
if group == "weekday":
return row.day.weekday()
if group == "hour":
return row.local.hour
if group == "run":
return row.id
if group == "all":
return "All runs"
number = value(row, width["field"])
return math_floor(number / width["width"]) if valid(number) else None
def label(token):
if group in {"day", "week", "month", "year"}:
return token.strftime("%d %b %Y")
if group == "weekday":
return ("Monday", "Tuesday", "Wednesday", "Thursday", "Friday", "Saturday", "Sunday")[
token
]
if group == "hour":
return f"{token:02d}:00"
if group == "run":
row = next(row for row in current if row.id == token)
return row.local.strftime("%d %b %Y %H:%M") + (
f" · {row.title}" if q["source"] != "runs" else f" · #{row.id}"
)
if group == "bin":
unit = catalog[width["field"]][1]
return f"{token * width['width']:g}–{(token + 1) * width['width']:g} {unit}"
return token
current_groups, prior_groups = defaultdict(list), defaultdict(list)
for row in current:
token = bucket(row)
if token is not None:
current_groups[token].append(row)
for row in prior:
token = bucket(
row, start - prior_start if group in {"day", "week", "month", "year"} else timedelta()
)
if token is not None:
prior_groups[token].append(row)
if group in {"day", "week", "month", "year"}:
dates = set()
for n in range((end - start).days):
day = start + timedelta(days=n)
dates.add(
day
if group == "day"
else day - timedelta(days=day.weekday())
if group == "week"
else day.replace(day=1)
if group == "month"
else day.replace(month=1, day=1)
)
tokens = sorted(dates)
elif group == "weekday":
tokens = list(range(7))
elif group == "hour":
tokens = list(range(24))
elif group == "run":
tokens = [row.id for row in current]
elif group == "all":
tokens = ["All runs"]
else:
tokens = sorted(set(current_groups) | set(prior_groups))
if len(tokens) > 1000 and not q["limit"]:
raise QueryError(
"This produces more than 1,000 chart groups. Use a wider grouping, take, or limit."
)
def measure(rows, key):
return summarize(rows, key, q)[0]
if q["having"]:
tokens = [
token
for token in tokens
if all(
condition_matches(measure(current_groups[token], condition["field"]), condition)
for condition in q["having"]
)
]
if q["rank"]:
key, direction = q["rank"]
with_data = [token for token in tokens if valid(measure(current_groups[token], key))]
without = [token for token in tokens if not valid(measure(current_groups[token], key))]
tokens = (
sorted(
with_data,
key=lambda token: measure(current_groups[token], key),
reverse=direction == "desc",
)
+ without
)
if q["limit"]:
tokens = (
tokens[: q["limit"]]
if q["rank"] or group not in {"day", "week", "month", "year"}
else tokens[-q["limit"] :]
)
if len(tokens) > 1000:
raise QueryError("Use at most 1,000 displayed groups.")
if q["having"] or q["limit"] or group == "bin":
kept = {row.id for token in tokens for row in current_groups[token]}
current = [row for row in current if row.id in kept]
kept_prior = {row.id for token in tokens for row in prior_groups[token]}
prior = [row for row in prior if row.id in kept_prior]
descriptors = {key: describe(key, catalog, q["aggregates"]) for key in keys}
for key, title in q["labels"].items():
if key in descriptors:
descriptors[key] = (title, *descriptors[key][1:])
cards = []
for key in keys:
number, coverage = summarize(current, key, q)
before, _ = summarize(prior, key, q)
title, unit, definition = descriptors[key]
if unit == "runs" and (key == "count" or q["aggregates"].get(key, {}).get("function") == "count"):
unit = "runs" if q["reduce"] else q["source"]
cards.append(
{
"key": key,
"label": title,
"unit": unit,
"definition": definition,
"value": formatted(number, key, q),
"coverage": coverage,
"previous": formatted(before, key, q),
"delta": f"{number - before:+.2f}" if valid(number) and valid(before) else None,
}
)
labels = [label(token) for token in tokens]
evidence_groups = {
text: current_groups[token] for token, text in zip(tokens, labels, strict=True)
}
raw_evidence_groups = dict(evidence_groups)
series, groups = [], []
for token, text in zip(tokens, labels, strict=True):
rows = current_groups[token]
groups.append(
{
"label": text,
"count": len(rows),
"values": [formatted(measure(rows, key), key, q) for key in keys],
"raw_values": [measure(rows, key) for key in keys],
}
)
for key in keys:
def measurements(mapping, key=key):
results = []
for index, token in enumerate(tokens):
contributing = mapping[token]
if q["trend"] == "cumulative":
contributing = [
row for earlier in tokens[: index + 1] for row in mapping[earlier]
]
elif q["trend"] and q["trend"].startswith("rolling "):
size = int(q["trend"].split()[1])
contributing = [
row
for earlier in tokens[max(0, index - size + 1) : index + 1]
for row in mapping[earlier]
]
results.append(measure(contributing, key))
if mapping is current_groups and key == keys[0]:
evidence_groups[label(token)] = contributing
if q["trend"] in {"change", "percent_change"}:
original = list(results)
results = [None] + [
(now - before if q["trend"] == "change" else (now - before) / abs(before) * 100)
if valid(now) and valid(before) and (q["trend"] == "change" or before != 0)
else None
for before, now in pairwise(original)
]
if not original:
results = []
return results
title, unit, _ = descriptors[key]
if unit == "runs" and (key == "count" or q["aggregates"].get(key, {}).get("function") == "count"):
unit = "runs" if q["reduce"] else q["source"]
pace_axis = unit == "/km"
if q["trend"] == "percent_change":
unit, pace_axis = "%", False
elif q["trend"] == "change" and pace_axis:
unit, pace_axis = "sec/km", False
series.append(
{
"key": key,
"label": title,
"unit": unit,
"values": measurements(current_groups),
"previous": measurements(prior_groups) if q["compare"] and group != "run" else [],
"pace_axis": pace_axis,
}
)
chart = {
"labels": labels,
"series": series,
"type": q["chart"],
"group": group,
"trend": q["trend"],
"display": q["display"],
"reference": q["reference"],
"dates": [token.isoformat() for token in tokens] if group == "day" else [],
"source": q["source"],
"timezone": timezone,
"timestamps": [next(row.instant.timestamp() * 1000 for row in current if row.id == token)
for token in tokens] if group == "run" else [],
"coverage": [evidence_coverage(current_groups[token], q["source"]) for token in tokens],
}
if q["cohorts"]:
cohorts = [{"label": cohort["name"], "values": []} for cohort in q["cohorts"]]
unassigned = {"label": "Other / missing", "values": []}
for token in tokens:
partitions = [[] for _ in range(len(cohorts) + 1)]
for row in current_groups[token]:
selected = next(
(
i
for i, cohort in enumerate(q["cohorts"])
if matches(row, cohort["conditions"])
),
len(cohorts),
)
partitions[selected].append(row)
for partition, cohort in zip(partitions, [*cohorts, unassigned], strict=True):
cohort["values"].append(measure(partition, keys[0]))
chart["cohorts"] = [*cohorts, unassigned]
from .stats_insights import enrich
insights = enrich(
chart, q, keys, tokens, current_groups, history, included, start, measure, catalog
)
chart["drill_labels"] = []
for key, series in zip(keys, chart["series"], strict=True):
histogram = series.get("histogram")
if not histogram or not histogram["counts"]:
continue
field = q["aggregates"].get(key, {}).get("field") or (
"count" if key in q["aggregates"] else key
)
edges = histogram["edges"]
bins = [[] for _ in histogram["counts"]]
for row in current:
number = value(row, field)
if valid(number):
index = min(len(bins) - 1, int((number - edges[0]) / histogram["width"]))
bins[index].append(row)
histogram["buckets"] = []
for index in range(len(histogram["counts"])):
name = f"Distribution · {key} · {index + 1}: {edges[index]:g}–{edges[index + 1]:g} {catalog[field][1]}"
rows = bins[index]
evidence_groups[name] = rows
raw_evidence_groups[name] = rows
histogram["buckets"].append(name)
chart["drill_labels"].append(name)
correlation = None
if q["plot"]:
x, y = q["plot"]
points = lambda rows: [
{
"x": value(row, x),
"y": value(row, y),
"id": getattr(row, "activity_id", row.id),
"label": row.title + " · " + row.local.strftime("%d %b %Y"),
"color": value(row, q["color"]) if q["color"] else None,
"size": value(row, q["size"]) if q["size"] else None,
}
for row in rows
if valid(value(row, x)) and valid(value(row, y))
]
plotted = points(current)
if len(plotted) >= 2:
xs, ys = [p["x"] for p in plotted], [p["y"] for p in plotted]
if len(set(xs)) > 1 and len(set(ys)) > 1:
import statistics
correlation = statistics.correlation(xs, ys)
chart["scatter"] = {
"x": catalog[x],
"y": catalog[y],
"current": plotted,
"previous": points(prior) if q["compare"] else [],
"correlation": correlation,
"color": catalog[q["color"]] if q["color"] else None,
"size": catalog[q["size"]] if q["size"] else None,
}
evidence = current
if selected_bucket is not None:
text = (
selected_bucket.strftime("%d %b %Y")
if hasattr(selected_bucket, "strftime")
else str(selected_bucket)
)
chosen_groups = raw_evidence_groups if raw_group else evidence_groups
if text not in chosen_groups:
raise QueryError("Choose a group from the current chart.")
evidence = ordered(chosen_groups[text], q["sort"])
pages = max(1, (len(evidence) + 24) // 25)
if not 1 <= page <= pages:
raise QueryError("That results page does not exist.")
def evidence_row(row):
inputs = [
q["aggregates"].get(key, {}).get("field")
or ("count" if key in q["aggregates"] else key)
for key in keys
]
return {
"id": getattr(row, "activity_id", row.id),
"title": row.title,
"date": row.local.strftime("%d %b %Y %H:%M"),
"distance": f"{row.distance_m / 1000:.2f}",
"values": [formatted(value(row, field), field, q) for field in inputs],
"raw_values": [value(row, field) for field in inputs],
"activity_ids": getattr(row, "activity_ids", []),
"previous_run": {
"id": row.previous.id,
"title": row.previous.title,
"date": row.previous.local.strftime("%d %b %Y %H:%M"),
}
if row.previous
else None,
}
return {
"cards": cards,
"source": "matched runs" if q["reduce"] else q["source"],
"coverage": evidence_coverage(current, q["source"]),
"group_coverage": [evidence_coverage(current_groups[token], q["source"]) for token in tokens],
"chart": chart,
"insights": insights,
"groups": groups,
"evidence_columns": [
{"key": field, "label": catalog[field][0], "unit": catalog[field][1]}
for key in keys
for field in [
q["aggregates"].get(key, {}).get("field")
or ("count" if key in q["aggregates"] else key)
]
],
"count": len(current),
"prior_count": len(prior),
"distance": sum(row.distance_m for row in current) / 1000,
"rows": [
evidence_row(row)
for row in (evidence if export else evidence[(page - 1) * 25 : page * 25])
],
"start": start,
"end": end - timedelta(days=1),
"previous_start": prior_start,
"previous_end": prior_end - timedelta(days=1),
"page": page,
"pages": pages,
"evidence_count": len(evidence),
"show_previous": any(
key in {"gap_since_previous_run", "distance_change", "pace_change", "heart_rate_change"}
or any(
name in key
for name in (
"gap_since_previous_run",
"distance_change",
"pace_change",
"heart_rate_change",
)
)
for key in keys
),
}
def math_floor(value):
import math
return math.floor(value)