K4LSH1_OPS: multi-coin LLM bot — DOGE/SOL/ETH, two-tier qwen+ornith, self-adapting
This commit is contained in:
drjones
279
test_bot.py
Normal file
279
test_bot.py
Normal file
@@ -0,0 +1,279 @@
|
||||
#!/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)
|
||||
Reference in New Issue
Block a user