141 lines
4.5 KiB
Python
141 lines
4.5 KiB
Python
#!/usr/bin/env python3
|
|
import json
|
|
import sqlite3
|
|
import time
|
|
from pathlib import Path
|
|
|
|
import requests
|
|
|
|
ROOT = Path('/home/drjones/.openclaw/workspace/alpaca-llm-bot-v1')
|
|
DB = ROOT / 'bot.db'
|
|
BASE = 'http://127.0.0.1:8089'
|
|
|
|
results = []
|
|
|
|
def record(name, ok, detail=''):
|
|
results.append({'test': name, 'ok': bool(ok), 'detail': str(detail)[:400]})
|
|
|
|
def test_health():
|
|
r = requests.get(f'{BASE}/health', timeout=5)
|
|
record('health_endpoint', r.ok and r.json().get('ok') is True, r.text[:120])
|
|
|
|
def test_metrics_schema():
|
|
r = requests.get(f'{BASE}/api/metrics?hours=24', timeout=8)
|
|
ok = r.ok
|
|
detail = ''
|
|
if ok:
|
|
j = r.json()
|
|
ok = j.get('ok') is True and 'decisionSeries' in j and 'activitySeries' in j and 'leaderboard' in j
|
|
detail = f"keys={list(j.keys())[:6]}"
|
|
record('metrics_endpoint_schema', ok, detail)
|
|
|
|
def _counts(cur):
|
|
out = {}
|
|
for t in ['insights','decisions','trades']:
|
|
cur.execute(f'select count(*) from {t}')
|
|
out[t] = cur.fetchone()[0]
|
|
return out
|
|
|
|
def test_curate_creates_insights():
|
|
con = sqlite3.connect(DB)
|
|
cur = con.cursor()
|
|
before = _counts(cur)['insights']
|
|
try:
|
|
requests.post(f'{BASE}/curate-now', timeout=25)
|
|
except Exception:
|
|
pass
|
|
time.sleep(1)
|
|
after = _counts(cur)['insights']
|
|
record('curate_increases_insights_or_stays', after >= before, f'before={before} after={after}')
|
|
con.close()
|
|
|
|
def test_run_now_creates_decision_activity():
|
|
con = sqlite3.connect(DB)
|
|
cur = con.cursor()
|
|
before = _counts(cur)['decisions']
|
|
try:
|
|
requests.post(f'{BASE}/run-now', timeout=35)
|
|
except Exception:
|
|
pass
|
|
time.sleep(1)
|
|
after = _counts(cur)['decisions']
|
|
record('run_now_increases_decisions_or_stays', after >= before, f'before={before} after={after}')
|
|
con.close()
|
|
|
|
def test_decision_integrity():
|
|
con = sqlite3.connect(DB)
|
|
cur = con.cursor()
|
|
cur.execute("select count(*) from decisions where symbol is null or symbol='' or action is null or status is null")
|
|
bad = cur.fetchone()[0]
|
|
record('decision_integrity_nonnull_fields', bad == 0, f'bad_rows={bad}')
|
|
con.close()
|
|
|
|
def test_notional_cap():
|
|
from config import settings
|
|
con = sqlite3.connect(DB)
|
|
cur = con.cursor()
|
|
cur.execute('select max(notional) from trades')
|
|
mx = cur.fetchone()[0] or 0
|
|
record('trade_notional_within_cap', mx <= settings.max_order_usd + 1e-9, f'max_notional={mx} cap={settings.max_order_usd}')
|
|
con.close()
|
|
|
|
def test_status_values():
|
|
allowed = {'planned','executed','failed','risk_blocked','skipped'}
|
|
con = sqlite3.connect(DB)
|
|
cur = con.cursor()
|
|
cur.execute('select distinct status from decisions order by status')
|
|
vals = {r[0] for r in cur.fetchall()}
|
|
ok = vals.issubset(allowed)
|
|
record('decision_status_enum', ok, f'statuses={sorted(vals)}')
|
|
con.close()
|
|
|
|
def test_ollama_tags():
|
|
from config import settings
|
|
try:
|
|
r = requests.get(settings.ollama_url + '/api/tags', timeout=8)
|
|
ok = r.ok and isinstance(r.json().get('models', []), list)
|
|
record('ollama_tags_reachable', ok, f"models={len(r.json().get('models',[])) if r.ok else 'n/a'}")
|
|
except Exception as e:
|
|
record('ollama_tags_reachable', False, str(e))
|
|
|
|
def test_qdrant_reachable():
|
|
from config import settings
|
|
try:
|
|
r = requests.get(settings.qdrant_url + '/collections', timeout=8)
|
|
ok = r.ok
|
|
record('qdrant_reachable', ok, r.text[:120])
|
|
except Exception as e:
|
|
record('qdrant_reachable', False, str(e))
|
|
|
|
def test_model_config_split():
|
|
from config import settings
|
|
ok = bool(settings.ollama_curator_model) and bool(settings.ollama_decision_model)
|
|
record('model_split_configured', ok, f"curator={settings.ollama_curator_model} decision={settings.ollama_decision_model}")
|
|
|
|
|
|
def main():
|
|
# import local config path
|
|
import sys
|
|
sys.path.insert(0, str(ROOT))
|
|
|
|
test_health()
|
|
test_metrics_schema()
|
|
test_curate_creates_insights()
|
|
test_run_now_creates_decision_activity()
|
|
test_decision_integrity()
|
|
test_notional_cap()
|
|
test_status_values()
|
|
test_ollama_tags()
|
|
test_qdrant_reachable()
|
|
test_model_config_split()
|
|
|
|
passed = sum(1 for r in results if r['ok'])
|
|
total = len(results)
|
|
summary = {'passed': passed, 'total': total, 'results': results, 'ts': int(time.time())}
|
|
out = ROOT / 'logs_flow_tests.json'
|
|
out.write_text(json.dumps(summary, indent=2))
|
|
print(json.dumps(summary, indent=2))
|
|
|
|
if __name__ == '__main__':
|
|
main()
|