"""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) if q["source"] in {"days", "health"}: for days in (7, 28): for field in ("distance", "count"): key = f"rolling_{days}d_{field}" label, unit, _ = catalog[key] catalog[key] = (label, unit, f"Complete recorded running {field} over this local day and the preceding {days - 1} local days, before filters. Rest days are included.") for spec in q["derived"]: catalog[spec["name"]] = ( 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"} or raw == "effort_time" or raw.startswith("best_"): 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, windows, totals = [], {7: deque(), 28: deque()}, {7: 0.0, 28: 0.0} previous, last_day, streak = None, None, 0 trees = { spec["name"]: expression_tree(spec["expression"], catalog)[0] for spec in q["derived"] if "window" not in spec } custom_windows = {spec["name"]: deque() for spec in q["derived"] if "window" in spec} 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 km in (1, 2, 5, 10): efforts = [e.get("seconds") for e in row.best_efforts or [] if e.get("distance") == km * 1000 and valid(e.get("seconds"))] row.values[f"best_{km}km"] = min(efforts) if efforts else None 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 for days in (7, 28): window = windows[days] while window and (row.instant - window[0].instant).total_seconds() >= days * 86400: totals[days] -= window.popleft().distance_m / 1000 window.append(row) totals[days] += row.distance_m / 1000 row.values[f"rolling_{days}d_distance"] = max(0.0, totals[days]) row.values[f"rolling_{days}d_count"] = len(window) 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 "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 != "sessions" and not key.startswith("rolling_"): 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 != "sessions" and not key.startswith("rolling_"): 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)