Issue 1341 — RecessionAlert leading U.S.-stock indicator¶

from datetime import datetime
from pathlib import Path
import hashlib
import json
import sqlite3

root = Path.cwd()
while root != root.parent and not (root / 'data' / 'recessionalert.sqlite').exists():
    root = root.parent
spec = json.loads((root / 'research/findings/specs/recessionalert-leading-us-stocks.json').read_text(encoding='utf-8'))
manifest = json.loads((root / 'research/artifacts/recessionalert/manifest.json').read_text(encoding='utf-8'))
inventory = (root / 'research/findings/recessionalert_inventory.md').read_text(encoding='utf-8')
url = 'https://recessionalert.com/a-leading-indicator-for-u-s-stocks/'
page_sha = 'sha256:e5a324ca2890226d7f62553757130605a5335c6b98bd88e4a67d22b2ca76ea01'
manifest_page = next(item for item in manifest if item.get('redacted_url') == url)
db_path = root / 'data/recessionalert.sqlite'
connection = sqlite3.connect(f'file:{db_path}?mode=ro', uri=True)
row = connection.execute('SELECT redacted_url, fetched_at, status, content_type, sha256, content_length, body FROM pages WHERE url=?', (url,)).fetchone()
history = connection.execute('SELECT fetched_at, status, kind, sha256 FROM crawl_log WHERE url=? ORDER BY fetched_at', (url,)).fetchall()
target_asset_count = connection.execute('SELECT COUNT(*) FROM assets WHERE page_url=? OR redacted_url=? OR url LIKE ?', (url, url, '%StocksLEI.gif%')).fetchone()[0]
all_asset_count = connection.execute('SELECT COUNT(*) FROM assets').fetchone()[0]
try:
    connection.execute('CREATE TABLE qm_1341_should_fail (id INTEGER)')
    raise AssertionError('read-only connection allowed DDL')
except sqlite3.OperationalError:
    pass
finally:
    connection.close()
redacted_url, fetched_at, status, content_type, digest, content_length, body = row
assert redacted_url == url
assert manifest_page['sha256'] == page_sha == digest == 'sha256:' + hashlib.sha256(body).hexdigest()
assert spec['source']['page']['sha256'] == page_sha
assert 'sha256:e5a324ca2' in inventory
assert content_length == len(body) == 83175
assert fetched_at == '2026-08-22T23:48:28Z'
assert history == [
    ('2026-08-22T23:36:04Z', 200, 'page', 'sha256:da85cd5c2915e4250431662982135fb48f477a87a6998d68b9c5ce27652f9d2a'),
    (fetched_at, 200, 'page', page_sha),
]
assert target_asset_count == 0
assert all_asset_count == 6
text = body.decode('utf-8', errors='replace')
for phrase in (
    'net percentage of 39 OECD countries with rising leading economic indices',
    'net percentage of 39 central banks that are easing rates',
    '7 and 10 months lead respectively',
    '0.57 r-square to NYSE',
    'buying opportunity',
    'StocksLEI.gif',
):
    assert phrase in text, phrase
assert spec['direction'] == 'buy_only_qualitative'
assert spec['output']['signal'] is None and spec['output']['playable'] is False
assert spec['fixture_expectations']['machine_readable_series_pinned'] is False
assert spec['fixture_expectations']['late_input_rejected'] is True
assert spec['fixture_expectations']['unavailable_state_preserved'] is True
assert spec['fixture_expectations']['registry_changed'] is False
article = datetime.fromisoformat('2024-05-01T15:35:21-04:00')
snapshot = datetime.fromisoformat(fetched_at.replace('Z', '+00:00'))
assert (snapshot - article).days == 843
print({
    'page_sha256': digest,
    'bytes': content_length,
    'fetched_at': fetched_at,
    'article_timestamp': spec['source']['page']['article_timestamp_disclosed'],
    'crawl_vintages': len(history),
    'canonical_assets': all_asset_count,
    'target_asset_rows': target_asset_count,
    'series_observations': 0,
    'source_claimed_leads_months': [7, 10],
    'source_claimed_r_squared': 0.57,
    'playable': spec['output']['playable'],
    'publication_lag': spec['publication_lag']['release_lag_status'],
    'disposition': spec['disposition'],
})