66 lines
2.1 KiB
Python
66 lines
2.1 KiB
Python
"""Smoke test for project B — run after build to verify wiring."""
|
|
import asyncio
|
|
import os
|
|
|
|
os.environ["TELEGRAM_ENABLED"] = "false"
|
|
os.environ["LOG_LEVEL"] = "INFO"
|
|
|
|
from src.db.database import CopyTraderDatabase
|
|
from src.services.kelly import KellySizer
|
|
from src.services.aggregator import SignalAggregator
|
|
|
|
# 1. Settings
|
|
from src.config import get_settings
|
|
s = get_settings()
|
|
print(f"[ok] settings: capital={s.initial_capital_usd} poll={s.user_poll_interval_seconds}s "
|
|
f"min_trade={s.min_trade_size_usd} consensus_wallets={s.consensus_min_wallets}")
|
|
|
|
# 2. DB
|
|
db = CopyTraderDatabase("data/test_smoke.db")
|
|
print(f"[ok] db initialized: {db.get_stats()}")
|
|
|
|
# 3. Kelly
|
|
sizer = KellySizer()
|
|
f = sizer.fraction(signal_strength=0.7, price=0.50, side="BUY")
|
|
size = sizer.position_usd(f, 10000)
|
|
print(f"[ok] kelly f(s=0.7, p=0.5, BUY)={f:.4f} size={size:.2f}")
|
|
|
|
# 4. Aggregator
|
|
agg = SignalAggregator(db, sizer=sizer)
|
|
agg.wallet_credibility["0xtest1"] = 0.8
|
|
print(f"[ok] aggregator ready")
|
|
|
|
|
|
async def test_signal_flow():
|
|
sample = {
|
|
"conditionId": "0xtest_market_abc",
|
|
"side": "BUY",
|
|
"outcome": "Yes",
|
|
"price": "0.55",
|
|
"size": "1000",
|
|
"usdcSize": "550",
|
|
}
|
|
# First trade — needs debounce + consensus
|
|
agg.wallet_credibility["0xtest1"] = 0.8
|
|
sig1 = await agg.on_trade("0xtest1", sample)
|
|
print(f"[ok] 1-wallet signal: {sig1} (expected None — needs CONSENSUS_MIN_WALLETS=2)")
|
|
|
|
# Second wallet
|
|
agg.wallet_credibility["0xtest2"] = 0.7
|
|
sig2 = await agg.on_trade("0xtest2", sample)
|
|
if sig2:
|
|
print(f"[ok] 2-wallet CONSENSUS REACHED:")
|
|
print(f" side={sig2['side']} outcome={sig2['outcome']} "
|
|
f"strength={sig2['aggregated_strength']:.3f} "
|
|
f"kelly={sig2['kelly_fraction']:.3f} "
|
|
f"size_usd={sig2['suggested_size_usd']:.2f} "
|
|
f"wallets={sig2['n_contributors']}")
|
|
else:
|
|
print(f"[warn] 2-wallet did not reach consensus — check threshold")
|
|
|
|
asyncio.run(test_signal_flow())
|
|
|
|
# Cleanup test db
|
|
os.remove("data/test_smoke.db")
|
|
print("\n=== Smoke test complete ===")
|