Issue 1361 — RecessionAlert CHARTS > Bitcoin¶

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-bitcoin.json').read_text(encoding='utf-8'))
page = spec['source']['page']
db = root / 'data' / 'recessionalert.sqlite'
connection = sqlite3.connect(f'file:{db}?mode=ro', uri=True)
connection.execute('PRAGMA query_only=ON')
row = connection.execute('SELECT sha256, content_length, fetched_at, body FROM pages WHERE url=?', (page['url'],)).fetchone()
target_asset_rows = connection.execute('SELECT COUNT(*) FROM assets WHERE page_url=? OR url=?', (page['url'], page['url'])).fetchone()[0]
crawl_count = connection.execute('SELECT COUNT(*) FROM crawl_log WHERE url=?', (page['url'],)).fetchone()[0]
connection.close()
assert row is not None
page_digest = 'sha256:' + hashlib.sha256(row[3]).hexdigest()
assert page_digest == row[0] == page['sha256']
assert row[1] == page['content_length'] == 116181
assert row[2] == page['fetched_at'] == '2026-08-22T23:39:03Z'
assert target_asset_rows == 0
assert crawl_count == len(spec['source']['crawl_revision_history']) == 5
workbook_root = root / 'research/artifacts/recessionalert'
for workbook in sorted(workbook_root.glob('*.xlsx')):
    digest = 'sha256:' + hashlib.sha256(workbook.read_bytes()).hexdigest()
    assert digest == next(item for item in spec['source']['adjacent_workbook_assets']['sha256'] if item.removeprefix('sha256:') == workbook.stem)
    print('workbook_sha256', workbook.name, digest)
print('page_sha256', page_digest)
print('page_content_length', row[1])
print('page_fetched_at', row[2])
print('target_asset_rows', target_asset_rows)
print('same_url_crawl_vintages', crawl_count)
print('source_series_linked', spec['provenance']['adjacent_workbook_series_linked'])
print('article_timestamp', spec['publication_lag']['article_timestamp'])
print('publication_lag', spec['publication_lag']['release_lag_status'])
print('decision', spec['disposition'])