280 lines
12 KiB
Python
280 lines
12 KiB
Python
#!/usr/bin/env python3
|
|
"""K4LSH1_OPS unit tests — verify every decision-making component."""
|
|
import json, sqlite3, sys, time, tempfile
|
|
from pathlib import Path
|
|
from unittest.mock import patch, MagicMock
|
|
|
|
sys.path.insert(0, str(Path(__file__).parent))
|
|
from bot import (momentum, decide, rsi, hedge_positions, llm_vote,
|
|
Kalshi, load_cfg, learned_min_conf, KRAKEN_PAIRS,
|
|
btc_price, intel)
|
|
|
|
PASS = 0
|
|
FAIL = 0
|
|
|
|
def check(name, condition, detail=""):
|
|
global PASS, FAIL
|
|
if condition:
|
|
PASS += 1
|
|
print(f" ✓ {name}")
|
|
else:
|
|
FAIL += 1
|
|
print(f" ✗ {name} — {detail}")
|
|
|
|
def test_momentum():
|
|
print("\n═══ MOMENTUM ═══")
|
|
conn = sqlite3.connect(":memory:")
|
|
conn.execute("CREATE TABLE ticks (ts REAL, price REAL)")
|
|
now = time.time()
|
|
# flat market
|
|
for i in range(20):
|
|
conn.execute("INSERT INTO ticks VALUES (?, 100.0)", (now - 300 + i*15,))
|
|
mom, score = momentum(conn)
|
|
check("flat market → FLAT", mom == "FLAT", f"got {mom}/{score:.4f}")
|
|
check("flat score near 0", score < 0.01, f"score={score:.4f}")
|
|
|
|
# up market
|
|
conn.execute("DELETE FROM ticks")
|
|
for i in range(20):
|
|
p = 100.0 + i * 0.5
|
|
conn.execute("INSERT INTO ticks VALUES (?,?)", (now - 300 + i*15, p))
|
|
mom, score = momentum(conn)
|
|
check("rising market → UP", mom == "UP", f"got {mom}/{score:.3f}%")
|
|
check("up score positive", score > 0.05, f"score={score:.3f}%")
|
|
|
|
# down market
|
|
conn.execute("DELETE FROM ticks")
|
|
for i in range(20):
|
|
p = 200.0 - i * 0.3
|
|
conn.execute("INSERT INTO ticks VALUES (?,?)", (now - 300 + i*15, p))
|
|
mom, score = momentum(conn)
|
|
check("falling market → DOWN", mom == "DOWN", f"got {mom}/{score:.3f}%")
|
|
check("down score positive", score > 0.01, f"score={score:.3f}%")
|
|
|
|
# too few ticks
|
|
conn.execute("DELETE FROM ticks")
|
|
conn.execute("INSERT INTO ticks VALUES (?, 100)", (now,))
|
|
mom, score = momentum(conn)
|
|
check("1 tick → FLAT", mom == "FLAT", f"got {mom}")
|
|
check("1 tick score 0", score == 0.0, f"score={score}")
|
|
conn.close()
|
|
|
|
|
|
def test_rsi():
|
|
print("\n═══ RSI ═══")
|
|
conn = sqlite3.connect(":memory:")
|
|
conn.execute("CREATE TABLE ticks (ts REAL, price REAL)")
|
|
now = time.time()
|
|
# oversold: steady decline
|
|
for i in range(30):
|
|
p = 100.0 - i * 0.5
|
|
conn.execute("INSERT INTO ticks VALUES (?,?)", (now - 1800 + i*60, p))
|
|
val = rsi(conn)
|
|
check("decline → RSI < 30", val is not None and val < 30, f"RSI={val}")
|
|
# overbought: steady climb
|
|
conn.execute("DELETE FROM ticks")
|
|
for i in range(30):
|
|
p = 100.0 + i * 0.8
|
|
conn.execute("INSERT INTO ticks VALUES (?,?)", (now - 1800 + i*60, p))
|
|
val = rsi(conn)
|
|
check("climb → RSI > 70", val is not None and val > 70, f"RSI={val}")
|
|
conn.close()
|
|
|
|
|
|
def test_learned_min_conf():
|
|
print("\n═══ ADAPTIVE CONFIDENCE ═══")
|
|
conn = sqlite3.connect(":memory:")
|
|
conn.execute("CREATE TABLE orders (id INTEGER PRIMARY KEY, pnl_cents INT, status TEXT)")
|
|
conn.execute("CREATE TABLE IF NOT EXISTS kv (k TEXT PRIMARY KEY, v REAL)")
|
|
cfg = {"learning": True, "min_conf": 0.55}
|
|
# < 8 resolved bets → use default
|
|
for i in range(5):
|
|
conn.execute("INSERT INTO orders(pnl_cents,status) VALUES (?,?)", (10, "won"))
|
|
lc = learned_min_conf(conn, cfg)
|
|
check("<8 bets → default", lc == 0.55, f"got {lc}")
|
|
|
|
# majority wins → relax confidence
|
|
for i in range(15):
|
|
conn.execute("INSERT INTO orders(pnl_cents,status) VALUES (?,?)", (15 if i < 10 else -10, "won" if i < 10 else "lost"))
|
|
lc = learned_min_conf(conn, cfg)
|
|
check("high WR → relax", lc < 0.55, f"got {lc}")
|
|
|
|
# majority losses → tighten (need fresh conn, kv starts empty)
|
|
conn.execute("DELETE FROM orders")
|
|
conn.execute("DELETE FROM kv")
|
|
for i in range(15):
|
|
conn.execute("INSERT INTO orders(pnl_cents,status) VALUES (?,?)",
|
|
(-10 if i < 10 else 15, "lost" if i < 10 else "won"))
|
|
lc = learned_min_conf(conn, {"learning": True, "min_conf": 0.55})
|
|
check("low WR → tighten", lc > 0.55, f"got {lc}")
|
|
conn.close()
|
|
|
|
|
|
def test_hedge():
|
|
print("\n═══ HEDGE LOGIC ═══")
|
|
conn = sqlite3.connect(":memory:")
|
|
conn.execute("CREATE TABLE orders (id INTEGER PRIMARY KEY, ts REAL, ticker TEXT, "
|
|
"side TEXT, count INT, price_cents INT, dry INT, order_id TEXT, status TEXT, "
|
|
"settle_ts REAL, pnl_cents INT)")
|
|
conn.execute("CREATE TABLE events (ts REAL, level TEXT, msg TEXT)")
|
|
cfg = {"dry_run": True, "min_profit_cents": 8}
|
|
ticker = "KXBTC15M-TEST"
|
|
book = {"yes": [[42]], "no": [[48]]}
|
|
|
|
# scenario: we bought NO at 45¢, YES ask is 42¢ → profit = 100-45-42=13¢ > 8¢ min
|
|
conn.execute("INSERT INTO orders VALUES (1,0,?,?,?,?,0,'sim','posted',NULL,NULL)",
|
|
(ticker, "no", 1, 45))
|
|
result = hedge_positions(conn, MagicMock(), ticker, cfg, book)
|
|
check("NO@45 YES@42 → hedge fires", result == "hedged", f"got {result}")
|
|
hedged = conn.execute("SELECT status FROM orders WHERE id=1").fetchone()[0]
|
|
check("original marked hedged", hedged == "hedged", f"got {hedged}")
|
|
new_orders = conn.execute("SELECT COUNT(*) FROM orders WHERE side='yes'").fetchone()[0]
|
|
check("counter-order placed", new_orders == 1)
|
|
|
|
# scenario: profit too small
|
|
conn.execute("DELETE FROM orders")
|
|
conn.execute("INSERT INTO orders VALUES (2,0,?,?,?,?,0,'sim','posted',NULL,NULL)",
|
|
(ticker, "no", 1, 48))
|
|
book2 = {"yes": [[49]], "no": [[50]]}
|
|
result2 = hedge_positions(conn, MagicMock(), ticker, cfg, book2)
|
|
check("NO@48 YES@49 → no hedge (1¢ profit)", result2 is None)
|
|
conn.close()
|
|
|
|
|
|
def test_decision_logic():
|
|
print("\n═══ DECISION LOGIC ═══")
|
|
conn = sqlite3.connect(":memory:")
|
|
conn.execute("CREATE TABLE ticks (ts REAL, price REAL)")
|
|
conn.execute("CREATE TABLE decisions (id INTEGER PRIMARY KEY, ts REAL, ticker TEXT, "
|
|
"mode TEXT, momentum TEXT, llm_vote TEXT, llm_conf REAL, llm_why TEXT, "
|
|
"final TEXT, price REAL, reason TEXT)")
|
|
conn.execute("CREATE TABLE orders (id INTEGER PRIMARY KEY, pnl_cents INT, status TEXT)")
|
|
conn.execute("CREATE TABLE events (ts REAL, level TEXT, msg TEXT)")
|
|
conn.execute("CREATE TABLE IF NOT EXISTS kv (k TEXT PRIMARY KEY, v REAL)")
|
|
now = time.time()
|
|
for i in range(20):
|
|
p = 100.0 + i * 0.3
|
|
conn.execute("INSERT INTO ticks VALUES (?,?)", (now - 300 + i*15, p))
|
|
|
|
cfg = {"swing_threshold_pct": 0.005, "min_conf": 0.50, "learning": False,
|
|
"mode": "AUTO", "coin": "BTC", "ollama_url": "http://x", "ollama_model": "x"}
|
|
|
|
# LLM confirms: UP spike, LLM votes DOWN (fade confirmed)
|
|
with patch("bot.intel", return_value={"chg24": 0.85, "fng": 45, "fee_fast": 8}), \
|
|
patch("bot.llm_vote", return_value=("DOWN", 0.72, "strong fade signal")):
|
|
mom, lv, lc, lw, final, why = decide(cfg, conn, 63400)
|
|
check("LLM fade confirm → DOWN", final == "DOWN", f"got {final}")
|
|
check("confidence preserved", lc == 0.72, f"got {lc}")
|
|
|
|
# LLM passes (not enough signal)
|
|
with patch("bot.intel", return_value={"chg24": 0.1}), \
|
|
patch("bot.llm_vote", return_value=("SKIP", 0.30, "no clear signal")):
|
|
mom, lv, lc, lw, final, why = decide(cfg, conn, 63400)
|
|
check("LLM skip → SKIP", final == "SKIP", f"got {final}")
|
|
|
|
# LLM goes against fade but high conf — should follow LLM
|
|
with patch("bot.intel", return_value={"chg24": 0.5}), \
|
|
patch("bot.llm_vote", return_value=("UP", 0.82, "continuation strong")):
|
|
mom, lv, lc, lw, final, why = decide(cfg, conn, 63400)
|
|
check("LLM continuation with high conf → trust LLM", final == "UP", f"got {final}/{why}")
|
|
|
|
# Low confidence → skip
|
|
with patch("bot.intel", return_value={"chg24": 0.3}), \
|
|
patch("bot.llm_vote", return_value=("DOWN", 0.35, "weak")):
|
|
mom, lv, lc, lw, final, why = decide(cfg, conn, 63400)
|
|
check("low conf → SKIP", final == "SKIP", f"got {final}")
|
|
|
|
conn.close()
|
|
|
|
|
|
def test_price_sources():
|
|
print("\n═══ PRICE SOURCES ═══")
|
|
for coin, pair in [("BTC","XBTUSD"), ("DOGE","XDGUSD"), ("SOL","SOLUSD"), ("ETH","ETHUSD")]:
|
|
p = btc_price(coin)
|
|
check(f"{coin} price from Kraken", p is not None and p > 0, f"${p}" if p else "None")
|
|
# coinbase fallback
|
|
with patch("bot.requests.get") as mock_get:
|
|
mock_kraken = MagicMock()
|
|
mock_kraken.json.side_effect = Exception("kraken down")
|
|
mock_cb = MagicMock()
|
|
mock_cb.json.return_value = {"data": {"amount": "99999"}}
|
|
mock_get.side_effect = [mock_kraken, mock_cb]
|
|
p = btc_price("BTC")
|
|
check("coinbase fallback works", p == 99999.0, f"got {p}")
|
|
|
|
|
|
def test_llm_vote_parsing():
|
|
print("\n═══ LLM VOTE PARSING ═══")
|
|
cfg = {"ollama_url": "http://x", "ollama_model": "x", "coin": "BTC"}
|
|
# ornith says DOWN with confidence
|
|
mock_resp = MagicMock()
|
|
mock_resp.text = json.dumps({"response": "DOWN|0.72|RSI oversold, fading up spike"})
|
|
with patch("bot.requests.post", return_value=mock_resp):
|
|
with patch("bot.intel", return_value={"chg24": 0.5}):
|
|
with patch("bot.rsi", return_value=28):
|
|
lv, lc, lw = llm_vote(cfg, 63400, "UP", 0.12, sqlite3.connect(":memory:"))
|
|
check("parses DOWN", lv == "DOWN", f"got {lv}")
|
|
check("parses confidence", lc == 0.72, f"got {lc}")
|
|
check("parses why", len(lw) > 5, f"'{lw}'")
|
|
|
|
# ornith says SKIP
|
|
mock_resp.text = json.dumps({"response": "SKIP|0.40|no clear direction"})
|
|
with patch("bot.requests.post", return_value=mock_resp):
|
|
with patch("bot.intel", return_value={"chg24": 0.1}):
|
|
with patch("bot.rsi", return_value=55):
|
|
lv, lc, lw = llm_vote(cfg, 63400, "FLAT", 0.001, sqlite3.connect(":memory:"))
|
|
check("parses SKIP", lv == "SKIP", f"got {lv}")
|
|
|
|
# malformed response
|
|
mock_resp.text = json.dumps({"response": "garbage noise no pipe"})
|
|
with patch("bot.requests.post", return_value=mock_resp):
|
|
with patch("bot.intel", return_value={}):
|
|
with patch("bot.rsi", return_value=None):
|
|
lv, lc, lw = llm_vote(cfg, 63400, "UP", 0.10, sqlite3.connect(":memory:"))
|
|
check("malformed → SKIP", lv == "SKIP", f"got {lv}")
|
|
check("malformed → 0 conf", lc == 0.0, f"got {lc}")
|
|
|
|
|
|
def test_config():
|
|
print("\n═══ CONFIG ═══")
|
|
with tempfile.NamedTemporaryFile(mode='w', suffix='.json', delete=False) as f:
|
|
json.dump({"mode": "AUTO", "stake_cents": 15, "coin": "TEST"}, f)
|
|
fpath = f.name
|
|
try:
|
|
from bot import HOME, CFG_PATH, load_cfg
|
|
old = str(CFG_PATH)
|
|
import bot
|
|
bot.CFG_PATH = Path(fpath)
|
|
c = load_cfg()
|
|
check("loads mode", c["mode"] == "AUTO")
|
|
check("loads stake", c["stake_cents"] == 15)
|
|
bot.CFG_PATH = Path(old)
|
|
finally:
|
|
Path(fpath).unlink(missing_ok=True)
|
|
|
|
|
|
def test_kraken_pairs():
|
|
print("\n═══ KRAKEN PAIRS ═══")
|
|
check("BTC → XBTUSD", KRAKEN_PAIRS["BTC"] == "XBTUSD")
|
|
check("DOGE → XDGUSD", KRAKEN_PAIRS["DOGE"] == "XDGUSD")
|
|
check("SOL → SOLUSD", KRAKEN_PAIRS["SOL"] == "SOLUSD")
|
|
check("ETH → ETHUSD", KRAKEN_PAIRS["ETH"] == "ETHUSD")
|
|
check("XRP → XRPUSD", KRAKEN_PAIRS["XRP"] == "XRPUSD")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
print("K4LSH1_OPS UNIT TESTS")
|
|
print("=" * 50)
|
|
tests = [test_momentum, test_rsi, test_learned_min_conf, test_hedge,
|
|
test_decision_logic, test_price_sources, test_llm_vote_parsing,
|
|
test_config, test_kraken_pairs]
|
|
for t in tests:
|
|
try:
|
|
t()
|
|
except Exception as e:
|
|
print(f" ✗ CRASH: {e}")
|
|
FAIL += 1
|
|
print(f"\n{'='*50}")
|
|
print(f"PASS: {PASS} FAIL: {FAIL}")
|
|
sys.exit(0 if FAIL == 0 else 1)
|