Viney Claude Sonnet 4.6 commited on
Commit
3ecd145
·
1 Parent(s): b002600

feat: add multi-quarter lexicon trend detection to textdiff

Browse files

Add `_detect_trend(counts)` pure helper that finds the longest strictly
monotone tail run (≥3 quarters) in a count series, and
`compute_lexicon_trend(ticker, n=4)` that applies it across all 18
_LEXICON terms over up to 4 recent 10-Q periods.

Wire into `_compute_inner` as step 5: trend signals for a term
supersede any QoQ term_frequency delta for the same term, so the
analyst never sees a redundant single-quarter spike alongside the
stronger multi-quarter narrative.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

Files changed (1) hide show
  1. analysis/textdiff.py +159 -0
analysis/textdiff.py CHANGED
@@ -356,6 +356,143 @@ def _find_context_sentence(text: str, pattern: str) -> str:
356
  return ""
357
 
358
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
359
  # ---------------------------------------------------------------------------
360
  # Guidance / MD&A language shift
361
  # ---------------------------------------------------------------------------
@@ -568,6 +705,28 @@ def _compute_inner(ticker: str, current_period: Optional[str]) -> list[QuarterDe
568
  if cur_mda and pri_mda:
569
  all_deltas.extend(compute_kpi_drops(cur_mda, pri_mda, period_from, period_to, form_type))
570
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
571
  # Deduplicate and sort: HIGH first, then MEDIUM, then LOW
572
  seen: set[str] = set()
573
  deduped: list[QuarterDelta] = []
 
356
  return ""
357
 
358
 
359
+ # ---------------------------------------------------------------------------
360
+ # Multi-quarter lexicon trend detection
361
+ # ---------------------------------------------------------------------------
362
+
363
+ def _detect_trend(counts: list[int]) -> str | None:
364
+ """Detect the longest strictly monotone run at the tail of a count series.
365
+
366
+ Args:
367
+ counts: term occurrence counts in **chronological order** (oldest first).
368
+
369
+ Returns:
370
+ ``"rising N quarters"`` if the last N≥3 values are strictly increasing,
371
+ ``"falling N quarters"`` if the last N≥3 values are strictly decreasing,
372
+ ``None`` otherwise.
373
+
374
+ Examples:
375
+ >>> _detect_trend([1, 3, 5, 8])
376
+ 'rising 4 quarters'
377
+ >>> _detect_trend([8, 5, 3, 1])
378
+ 'falling 4 quarters'
379
+ >>> _detect_trend([1, 5, 2, 4, 6])
380
+ 'rising 3 quarters'
381
+ >>> _detect_trend([1, 2, 2, 4]) # plateau breaks strict run
382
+ >>> _detect_trend([1, 3]) # only 2 values
383
+ """
384
+ if len(counts) < 2:
385
+ return None
386
+
387
+ # Walk backwards from the end to find the longest tail run
388
+ # We track whether the tail is rising or falling from the last step
389
+ n = len(counts)
390
+
391
+ # Determine direction of the final step
392
+ if counts[-1] > counts[-2]:
393
+ direction = "rising"
394
+ elif counts[-1] < counts[-2]:
395
+ direction = "falling"
396
+ else:
397
+ return None # last step is flat → no strict run
398
+
399
+ # Extend the run backwards as far as the same strict direction holds
400
+ run_length = 2 # we already know the last pair qualifies
401
+ for i in range(n - 2, 0, -1):
402
+ if direction == "rising" and counts[i] > counts[i - 1]:
403
+ run_length += 1
404
+ elif direction == "falling" and counts[i] < counts[i - 1]:
405
+ run_length += 1
406
+ else:
407
+ break # run ends here
408
+
409
+ if run_length < 3:
410
+ return None
411
+
412
+ return f"{direction} {run_length} quarters"
413
+
414
+
415
+ def compute_lexicon_trend(ticker: str, n: int = 4) -> list[QuarterDelta]:
416
+ """Detect multi-quarter monotone trends for each analyst-lexicon term.
417
+
418
+ Looks back up to *n* 10-Q periods and surfaces terms whose occurrence
419
+ count has been strictly rising or falling for 3+ consecutive quarters —
420
+ a more durable signal than a single quarter-over-quarter spike.
421
+
422
+ Args:
423
+ ticker: uppercase ticker symbol.
424
+ n: maximum number of recent 10-Q periods to examine (default 4).
425
+
426
+ Returns:
427
+ Up to 5 ``QuarterDelta`` objects (HIGH-significance first), one per
428
+ term that shows a multi-quarter trend. Returns ``[]`` if fewer than
429
+ 3 periods are available or no trends are detected.
430
+ """
431
+ ticker = ticker.upper()
432
+ periods = get_periods_for_ticker(ticker, form_type="10-Q")
433
+ if len(periods) < 3:
434
+ return []
435
+
436
+ n = min(n, len(periods))
437
+ # periods[:n] is newest-first; reverse for chronological order
438
+ selected = list(reversed(periods[:n])) # [oldest, ..., newest]
439
+
440
+ # Pre-load section text for each period
441
+ period_texts: list[str] = []
442
+ for period in selected:
443
+ mda = get_section(ticker, period, "mda") or ""
444
+ risk = get_section(ticker, period, "risk_factors") or ""
445
+ period_texts.append((mda + "\n\n" + risk).strip())
446
+
447
+ deltas: list[QuarterDelta] = []
448
+
449
+ for pattern, label in _LEXICON:
450
+ counts = [
451
+ len(re.findall(pattern, text, re.IGNORECASE))
452
+ for text in period_texts
453
+ ]
454
+
455
+ trend = _detect_trend(counts)
456
+ if trend is None:
457
+ continue
458
+
459
+ # Build the QuarterDelta
460
+ oldest_period = selected[0] # periods[n-1] in newest-first order
461
+ newest_period = selected[-1] # periods[0]
462
+
463
+ first_count = counts[0]
464
+ last_count = counts[-1]
465
+ metric = (
466
+ f"{first_count}→{last_count} occurrences over "
467
+ f"{oldest_period}→{newest_period} ({trend})"
468
+ )
469
+
470
+ oldest_text = period_texts[0]
471
+ newest_text = period_texts[-1]
472
+ before_ctx = _find_context_sentence(oldest_text, pattern) if first_count > 0 else ""
473
+ after_ctx = _find_context_sentence(newest_text, pattern) if last_count > 0 else ""
474
+
475
+ # Significance: HIGH for runs of 4+, MEDIUM for 3
476
+ run_quarters = int(trend.split()[1])
477
+ sig = "HIGH" if run_quarters >= 4 else "MEDIUM"
478
+
479
+ deltas.append(QuarterDelta(
480
+ kind="term_frequency",
481
+ period_from=oldest_period,
482
+ period_to=newest_period,
483
+ before_text=before_ctx,
484
+ after_text=after_ctx,
485
+ computed_metric=metric,
486
+ source="10-Q",
487
+ significance=sig,
488
+ term=label,
489
+ ))
490
+
491
+ # HIGH first, then MEDIUM; cap at 5
492
+ deltas.sort(key=lambda d: {"HIGH": 0, "MEDIUM": 1}.get(d.significance, 2))
493
+ return deltas[:5]
494
+
495
+
496
  # ---------------------------------------------------------------------------
497
  # Guidance / MD&A language shift
498
  # ---------------------------------------------------------------------------
 
705
  if cur_mda and pri_mda:
706
  all_deltas.extend(compute_kpi_drops(cur_mda, pri_mda, period_from, period_to, form_type))
707
 
708
+ # 5. Multi-quarter lexicon trends (ticker-level, not period-pair)
709
+ trend_deltas = compute_lexicon_trend(ticker)
710
+ all_deltas.extend(trend_deltas)
711
+
712
+ # Prefer trend signals over QoQ signals for the same term:
713
+ # collect terms that have a multi-quarter trend signal and remove any
714
+ # plain QoQ term_frequency delta for the same term.
715
+ trend_terms: set[str] = {
716
+ d.term
717
+ for d in trend_deltas
718
+ if d.kind == "term_frequency" and "quarters" in (d.computed_metric or "")
719
+ }
720
+ if trend_terms:
721
+ all_deltas = [
722
+ d for d in all_deltas
723
+ if not (
724
+ d.kind == "term_frequency"
725
+ and d.term in trend_terms
726
+ and "quarters" not in (d.computed_metric or "")
727
+ )
728
+ ]
729
+
730
  # Deduplicate and sort: HIGH first, then MEDIUM, then LOW
731
  seen: set[str] = set()
732
  deduped: list[QuarterDelta] = []