diff --git a/crypto_sentiment_crawler/analysis/source_weights.py b/crypto_sentiment_crawler/analysis/source_weights.py index 29eed35..e643e03 100644 --- a/crypto_sentiment_crawler/analysis/source_weights.py +++ b/crypto_sentiment_crawler/analysis/source_weights.py @@ -151,7 +151,9 @@ def compute_weights_from_beliefs(beliefs: dict, min_samples: int = 20) -> dict: "is_contrarian": is_contrarian, "alpha": belief.get("alpha"), "beta": belief.get("beta"), - "sample_size": belief.get("effective_n", belief.get("alpha", 1) + belief.get("beta", 1)), + "sample_size": belief.get( + "effective_n", belief.get("alpha", 1) + belief.get("beta", 1) + ), } # Normalize weights to sum to 1.0 @@ -195,6 +197,46 @@ async def _save_weights_to_db( await create_weights_table(db) try: + if not weights: + raise ValueError("Refusing to stage or publish an empty source-weight snapshot") + if belief_version is not None and ( + type(belief_version) is not int or belief_version < 0 + ): + raise ValueError("belief_version must be a non-negative integer") + + # Serialize publishers before inspecting the currently accepted version. + # The JSON state file is published separately, so equal-version retries + # must be idempotent while stale writers must never prune newer rows. + await db.conn.execute("BEGIN IMMEDIATE") + publication = await ( + await db.conn.execute( + "SELECT belief_version FROM belief_publications WHERE id = 1" + ) + ).fetchone() + published_version = int(publication[0]) if publication is not None else None + if ( + belief_version is not None + and published_version is not None + and belief_version < published_version + ): + raise ValueError( + f"Refusing stale source-weight version {belief_version}; " + f"published version is {published_version}" + ) + + if belief_version is not None: + newer_current = await ( + await db.conn.execute( + "SELECT COUNT(*) FROM source_weights " + "WHERE belief_version IS NOT NULL AND belief_version > ?", + (belief_version,), + ) + ).fetchone() + if int(newer_current[0]): + raise RuntimeError( + "source_weights contains rows newer than the requested publication" + ) + if belief_version is not None: await db.conn.execute( "DELETE FROM source_weight_snapshots WHERE belief_version = ?", @@ -203,7 +245,8 @@ async def _save_weights_to_db( for source, data in weights.items(): await db.conn.execute(""" INSERT INTO source_weight_snapshots - (belief_version, source, weight, accuracy, is_contrarian, alpha, beta, sample_size, last_updated) + (belief_version, source, weight, accuracy, is_contrarian, + alpha, beta, sample_size, last_updated) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?) """, ( belief_version, @@ -217,11 +260,21 @@ async def _save_weights_to_db( datetime.now(timezone.utc).isoformat(), )) + staged = await ( + await db.conn.execute( + "SELECT COUNT(*) FROM source_weight_snapshots WHERE belief_version = ?", + (belief_version,), + ) + ).fetchone() + if int(staged[0]) != len(weights): + raise RuntimeError("Staged source-weight snapshot is incomplete") + if publish: for source, data in weights.items(): await db.conn.execute(""" INSERT INTO source_weights - (source, weight, accuracy, is_contrarian, alpha, beta, sample_size, belief_version, last_updated) + (source, weight, accuracy, is_contrarian, alpha, beta, + sample_size, belief_version, last_updated) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?) ON CONFLICT(source) DO UPDATE SET weight = excluded.weight, @@ -249,6 +302,47 @@ async def _save_weights_to_db( datetime.now(timezone.utc).isoformat(), )) if belief_version is not None: + # `source_weights` is the compatibility/current mirror. Remove + # sources absent from the complete accepted snapshot in the same + # transaction as the upserts and publication pointer update. + # Historical versions remain available in the snapshot table. + await db.conn.execute( + """ + DELETE FROM source_weights + WHERE NOT EXISTS ( + SELECT 1 + FROM source_weight_snapshots snapshot + WHERE snapshot.belief_version = ? + AND snapshot.source = source_weights.source + ) + """, + (belief_version,), + ) + + mirror = await ( + await db.conn.execute( + """ + SELECT COUNT(*) + FROM source_weights current + JOIN source_weight_snapshots snapshot + ON snapshot.belief_version = ? + AND snapshot.source = current.source + WHERE current.belief_version = ? + AND current.weight = snapshot.weight + AND current.accuracy IS snapshot.accuracy + AND current.is_contrarian = snapshot.is_contrarian + AND current.alpha IS snapshot.alpha + AND current.beta IS snapshot.beta + AND current.sample_size IS snapshot.sample_size + """, + (belief_version, belief_version), + ) + ).fetchone() + if int(mirror[0]) != len(weights): + raise RuntimeError( + "source_weights does not exactly mirror the staged snapshot" + ) + await db.conn.execute(""" INSERT INTO belief_publications (id, belief_version, published_at) VALUES (1, ?, ?) @@ -394,16 +488,28 @@ def print_weights_table(weights: dict): reverse=True ) - print(f"\n{'Source':<30} {'Weight':<10} {'Norm':<10} {'Accuracy':<10} {'Type':<12} {'Samples':<10}") + print( + f"\n{'Source':<30} {'Weight':<10} {'Norm':<10} " + f"{'Accuracy':<10} {'Type':<12} {'Samples':<10}" + ) print("-" * 80) for source, data in sorted_weights: accuracy = data.get("accuracy") acc_str = f"{accuracy:.1%}" if accuracy else "N/A" - type_str = "CONTRARIAN" if data.get("is_contrarian") else "MOMENTUM" if accuracy and accuracy > 0.5 else "NEUTRAL" + type_str = ( + "CONTRARIAN" + if data.get("is_contrarian") + else "MOMENTUM" + if accuracy and accuracy > 0.5 + else "NEUTRAL" + ) norm = data.get("weight_normalized", 0) samples = data.get("sample_size", 0) - print(f"{source:<30} {data['weight']:<10.4f} {norm:<10.4f} {acc_str:<10} {type_str:<12} {samples:<10.0f}") + print( + f"{source:<30} {data['weight']:<10.4f} {norm:<10.4f} " + f"{acc_str:<10} {type_str:<12} {samples:<10.0f}" + ) print("=" * 80) diff --git a/tests/test_state_snapshot_ownership.py b/tests/test_state_snapshot_ownership.py index 54077ae..b5f2427 100644 --- a/tests/test_state_snapshot_ownership.py +++ b/tests/test_state_snapshot_ownership.py @@ -601,20 +601,21 @@ async def test_weight_loader_ignores_stale_snapshot_versions(tmp_path: Path): }, belief_version=2, ) - await save_weights_to_db( - db, - { - "current_source": { - "weight": 0.8, - "accuracy": 0.8, - "is_contrarian": False, - "alpha": 8.0, - "beta": 2.0, - "sample_size": 10, - } - }, - belief_version=1, - ) + with pytest.raises(ValueError, match="Refusing stale source-weight version"): + await save_weights_to_db( + db, + { + "current_source": { + "weight": 0.8, + "accuracy": 0.8, + "is_contrarian": False, + "alpha": 8.0, + "beta": 2.0, + "sample_size": 10, + } + }, + belief_version=1, + ) finally: await db.close() @@ -661,3 +662,216 @@ async def test_weight_loader_retries_when_state_changes_during_read(tmp_path: Pa assert loaded["belief_version"] == 2 assert loaded["weights"] == {"new_source": 0.2} + + +def _weight(value: float) -> dict: + return { + "weight": value, + "accuracy": 0.6, + "is_contrarian": False, + "alpha": 3.0, + "beta": 2.0, + "sample_size": 5, + } + + +async def test_published_weights_exactly_mirror_a_shrinking_snapshot(tmp_path: Path): + db = Database(tmp_path / "sentiment.db") + await db.connect() + try: + await save_weights_to_db( + db, + {"keep": _weight(0.2), "remove": _weight(0.8)}, + belief_version=1, + ) + await save_weights_to_db(db, {"keep": _weight(0.4)}, belief_version=2) + + current = await ( + await db.conn.execute( + "SELECT source, weight, belief_version FROM source_weights ORDER BY source" + ) + ).fetchall() + active = await ( + await db.conn.execute( + "SELECT source, weight, belief_version FROM active_source_weights ORDER BY source" + ) + ).fetchall() + old_snapshot = await ( + await db.conn.execute( + "SELECT source FROM source_weight_snapshots " + "WHERE belief_version = 1 ORDER BY source" + ) + ).fetchall() + finally: + await db.close() + + assert [tuple(row) for row in current] == [("keep", 0.4, 2)] + assert [tuple(row) for row in active] == [("keep", 0.4, 2)] + assert [row["source"] for row in old_snapshot] == ["keep", "remove"] + + +async def test_equal_version_republish_is_exact_and_idempotent(tmp_path: Path): + db = Database(tmp_path / "sentiment.db") + await db.connect() + try: + await save_weights_to_db( + db, + {"old": _weight(0.2), "keep": _weight(0.3)}, + belief_version=4, + ) + await save_weights_to_db(db, {"keep": _weight(0.7)}, belief_version=4) + + current = await ( + await db.conn.execute( + "SELECT source, weight, belief_version FROM source_weights" + ) + ).fetchall() + snapshot = await ( + await db.conn.execute( + "SELECT source, weight, belief_version FROM source_weight_snapshots " + "WHERE belief_version = 4" + ) + ).fetchall() + publication = await ( + await db.conn.execute( + "SELECT belief_version FROM belief_publications WHERE id = 1" + ) + ).fetchone() + finally: + await db.close() + + assert [tuple(row) for row in current] == [("keep", 0.7, 4)] + assert [tuple(row) for row in snapshot] == [("keep", 0.7, 4)] + assert publication["belief_version"] == 4 + + +async def test_stale_weight_publication_is_rejected_without_changes(tmp_path: Path): + db = Database(tmp_path / "sentiment.db") + await db.connect() + try: + await save_weights_to_db(db, {"current": _weight(0.6)}, belief_version=2) + + with pytest.raises(ValueError, match="Refusing stale source-weight version"): + await save_weights_to_db(db, {"stale": _weight(0.9)}, belief_version=1) + + current = await ( + await db.conn.execute( + "SELECT source, weight, belief_version FROM source_weights" + ) + ).fetchall() + stale_snapshots = await ( + await db.conn.execute( + "SELECT COUNT(*) FROM source_weight_snapshots WHERE belief_version = 1" + ) + ).fetchone() + publication = await ( + await db.conn.execute( + "SELECT belief_version FROM belief_publications WHERE id = 1" + ) + ).fetchone() + finally: + await db.close() + + assert [tuple(row) for row in current] == [("current", 0.6, 2)] + assert stale_snapshots[0] == 0 + assert publication["belief_version"] == 2 + + +async def test_versioned_publish_prunes_legacy_null_weight_rows(tmp_path: Path): + db = Database(tmp_path / "sentiment.db") + await db.connect() + try: + await db.conn.execute( + "INSERT INTO source_weights (source, weight, belief_version) VALUES (?, ?, NULL)", + ("legacy", 0.9), + ) + await db.conn.commit() + + await save_weights_to_db(db, {"current": _weight(0.5)}, belief_version=1) + rows = await ( + await db.conn.execute( + "SELECT source, belief_version FROM source_weights ORDER BY source" + ) + ).fetchall() + finally: + await db.close() + + assert [tuple(row) for row in rows] == [("current", 1)] + + +async def test_empty_weight_snapshot_is_rejected_without_publication(tmp_path: Path): + db = Database(tmp_path / "sentiment.db") + await db.connect() + try: + await save_weights_to_db(db, {"current": _weight(0.5)}, belief_version=1) + + with pytest.raises(ValueError, match="empty source-weight snapshot"): + await save_weights_to_db(db, {}, belief_version=2, publish=False) + with pytest.raises(ValueError, match="empty source-weight snapshot"): + await save_weights_to_db(db, {}, belief_version=2) + + publication = await ( + await db.conn.execute( + "SELECT belief_version FROM belief_publications WHERE id = 1" + ) + ).fetchone() + staged = await ( + await db.conn.execute( + "SELECT COUNT(*) FROM source_weight_snapshots WHERE belief_version = 2" + ) + ).fetchone() + finally: + await db.close() + + assert publication["belief_version"] == 1 + assert staged[0] == 0 + + +async def test_weight_mirror_failure_rolls_back_snapshot_and_upserts(tmp_path: Path): + db = Database(tmp_path / "sentiment.db") + await db.connect() + try: + await save_weights_to_db( + db, + {"keep": _weight(0.2), "drop_me": _weight(0.8)}, + belief_version=1, + ) + await db.conn.execute( + """ + CREATE TRIGGER fail_weight_prune + BEFORE DELETE ON source_weights + WHEN OLD.source = 'drop_me' + BEGIN + SELECT RAISE(ABORT, 'injected weight-prune failure'); + END + """ + ) + await db.conn.commit() + + with pytest.raises(Exception, match="injected weight-prune failure"): + await save_weights_to_db(db, {"keep": _weight(0.7)}, belief_version=2) + + current = await ( + await db.conn.execute( + "SELECT source, weight, belief_version FROM source_weights ORDER BY source" + ) + ).fetchall() + staged = await ( + await db.conn.execute( + "SELECT COUNT(*) FROM source_weight_snapshots WHERE belief_version = 2" + ) + ).fetchone() + publication = await ( + await db.conn.execute( + "SELECT belief_version FROM belief_publications WHERE id = 1" + ) + ).fetchone() + finally: + await db.close() + + assert [tuple(row) for row in current] == [ + ("drop_me", 0.8, 1), + ("keep", 0.2, 1), + ] + assert staged[0] == 0 + assert publication["belief_version"] == 1