reporting_lag.py · Scientific Agent Skills

1#!/usr/bin/env python3

2# /// script

3# requires-python = ">=3.11"

4# dependencies = []

5# ///

6"""Describe collection-to-submission/release delays among currently visible records.

7

8This is a conditional empirical CDF, not an estimate of eventual completeness.

9Monthly cohorts are grouped by both dates. Partial collection dates, negative

10lags, missing dates and submissions after the analysis anchor are excluded.

11"""

12from __future__ import annotations

13

14import argparse

15import sys

16from datetime import date, timedelta

17

18from lapis_client import (

19 LapisError, aggregated, data_version, describe_instance, emit,

20 pick_date_field, range_keys, resolve_base_url, surveillance_filters,

21)

22

23COLUMNS = ("lag_days", "mean_observed", "min_observed", "max_observed", "cohorts")

24OFFSETS = (7, 14, 21, 30, 45, 60, 90, 120, 180)

25

26

27def month_window(anchor: date, months_back: int) -> tuple[date, date]:

28 """First and last day of the month ``months_back`` before ``anchor``."""

29 total = anchor.year * 12 + (anchor.month - 1) - months_back

30 year, month = divmod(total, 12)

31 first = date(year, month + 1, 1)

32 nxt = date(year + 1, 1, 1) if month == 11 else date(year, month + 2, 1)

33 return first, nxt - timedelta(days=1)

34

35

36def cohort_curve(base_url: str, filters: dict, submission_field: str,

37 collection_field: str, collection_upper: str | None = None,

38 anchor: date | None = None) -> tuple[dict[int, float], int, int] | None:

39 """Return (observed delay CDF, eligible count, excluded count) for a cohort.

40

41 A range-lower collection field requires a matching upper bound to retain

42 only exact dates. No month-end approximation or negative-lag clamping.

43 """

44 fields = [collection_field, submission_field]

45 if collection_upper:

46 fields.append(collection_upper)

47 rows = aggregated(base_url, filters, fields)

48 lags: list[tuple[int, int]] = []

49 dated = excluded = 0

50 for row in rows:

51 n = int(row.get("count") or 0)

52 try:

53 submitted = date.fromisoformat(str(row.get(submission_field) or ""))

54 collected = date.fromisoformat(str(row.get(collection_field) or ""))

55 except ValueError:

56 excluded += n

57 continue

58 if ((collection_upper and row.get(collection_upper) != row.get(collection_field))

59 or submitted < collected or (anchor and submitted > anchor)):

60 excluded += n

61 continue

62 lags.append(((submitted - collected).days, n))

63 dated += n

64 if not dated:

65 return ({}, 0, excluded) if excluded else None

66 curve = {offset: sum(n for lag, n in lags if lag <= offset) / dated for offset in OFFSETS}

67 return curve, dated, excluded

68

69

70def build_parser() -> argparse.ArgumentParser:

71 parser = argparse.ArgumentParser(description="Observed collection-to-release delays; not completeness.")

72 parser.add_argument("--instance", default="sars-cov-2")

73 parser.add_argument("--base-url", help="another trusted public LAPIS deployment")

74 parser.add_argument("--date-field", help="collection-date column")

75 parser.add_argument("--submission-field", help="submission/release-date column")

76 parser.add_argument("--where", action="append", default=[], metavar="KEY=VALUE")

77 parser.add_argument("--cohorts", type=int, default=6, help="monthly cohorts (default: 6)")

78 parser.add_argument("--skip-months", type=int, default=3, help="youngest cohort in months back (default: 3)")

79 parser.add_argument("--target", type=float, default=0.9, help="target observed CDF fraction (default: 0.9)")

80 parser.add_argument("--until", help="analysis anchor YYYY-MM-DD, not a historical database snapshot")

81 parser.add_argument("--format", choices=("table", "tsv", "json"), default="table")

82 return parser

83

84

85def main(argv: list[str] | None = None) -> int:

86 args = build_parser().parse_args(argv)

87 try:

88 if args.cohorts < 1 or args.skip_months < 1 or not 0 < args.target <= 1:

89 raise LapisError("--cohorts/--skip-months must be positive; --target must be in (0, 1]")

90 anchor = date.fromisoformat(args.until) if args.until else date.today()

91 base_url = resolve_base_url(args.instance, args.base_url)

92 schema = describe_instance(base_url)

93 collection = pick_date_field(schema, "collection", args.date_field)

94 submission = pick_date_field(schema, "submission", args.submission_field)

95 if collection == submission:

96 raise LapisError("collection and submission fields must differ")

97 upper = collection.replace("RangeLower", "RangeUpper") if collection.endswith("RangeLower") else None

98 if upper and upper not in schema["types"]:

99 raise LapisError("collection range lacks an upper bound; cannot identify exact dates")

100 where = {}

101 for pair in args.where:

102 key, sep, value = pair.partition("=")

103 if not sep or key.split(".")[0] not in schema["types"]:

104 raise LapisError(f"bad --where {pair!r}")

105 if key.split(".")[0] in {collection, submission, upper}:

106 raise LapisError("--where cannot replace collection/submission date selection")

107 where[key] = value

108 where = surveillance_filters(schema, where)

109 version = data_version(base_url)

110 from_key, to_key = range_keys(collection)

111 curves, sizes, excluded_total = [], [], 0

112 for back in range(args.skip_months, args.skip_months + args.cohorts):

113 start, end = month_window(anchor, back)

114 result = cohort_curve(base_url, {**where, from_key: start.isoformat(), to_key: end.isoformat()},

115 submission, collection, upper, anchor)

116 if result:

117 curve, dated, excluded = result

118 excluded_total += excluded

119 # Cohorts with shorter follow-up cannot contribute long offsets.

120 eligible = {o: v for o, v in curve.items() if o <= (anchor - end).days}

121 if eligible:

122 curves.append(eligible)

123 sizes.append(dated)

124 if not curves:

125 print(f"error: no eligible cohort; {excluded_total} records excluded for date quality", file=sys.stderr)

126 return 1

127 rows, means = [], {}

128 for offset in OFFSETS:

129 values = [c[offset] for c in curves if offset in c]

130 if not values:

131 continue

132 means[offset] = sum(values) / len(values)

133 rows.append({"lag_days": offset, "mean_observed": f"{means[offset]:.3f}",

134 "min_observed": f"{min(values):.3f}", "max_observed": f"{max(values):.3f}",

135 "cohorts": len(values)})

136 except (LapisError, ValueError) as exc:

137 print(f"error: {exc}", file=sys.stderr)

138 return 2

139 print(emit(rows, COLUMNS, args.format))

140 reached = [o for o, v in means.items() if v >= args.target]

141 print(f"\n# {schema['name']} via {base_url} | data version {version}"

142 f"\n# collection {collection}; submission/release {submission}; filters {where}"

143 f"\n# anchor {anchor}; cohorts {args.skip_months}..{args.skip_months + args.cohorts - 1} months back"

144 f"\n# {sum(sizes)} eligible records; {excluded_total} excluded: missing/imprecise/reversed/post-anchor dates"

145 "\n# Equal cohort weighting; cohort count may differ by offset, so the mean need not be monotonic.",

146 file=sys.stderr)

147 if reached:

148 cutoff = anchor - timedelta(days=reached[0])

149 print(f"# Mean observed CDF reaches {args.target:.0%} by {reached[0]} days; heuristic date {cutoff}."

150 "\n# This does not establish completeness or make any collection week trustworthy.", file=sys.stderr)

151 else:

152 print(f"# No observed offset reaches target {args.target:.0%}; no heuristic date reported.", file=sys.stderr)

153 print("# Conditional on records visible now: future submissions can lengthen delays."

154 "\n# Release date need not equal first appearance here; historical snapshots are not reconstructed.",

155 file=sys.stderr)

156 return 0

157

158

159if __name__ == "__main__":

160 sys.exit(main())