Tacit Lab
52cd76a750
Replace the V1 commands/services split with a flat, direct architecture: - cli.py dispatches directly to service functions - New services: account, market, trade, opportunity - Thin Binance wrappers: spot_client, um_futures_client - Add audit logging, runtime paths, and TOML config - Remove legacy V1 code: commands/, precheck, review engine, smart executor - Add ruff + mypy toolchain and fix edge cases in trade params Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
95 lines
4.3 KiB
Python
95 lines
4.3 KiB
Python
"""Opportunity service tests."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
from coinhunter.services import opportunity_service
|
|
|
|
|
|
class FakeSpotClient:
|
|
def account_info(self):
|
|
return {
|
|
"balances": [
|
|
{"asset": "USDT", "free": "50", "locked": "0"},
|
|
{"asset": "BTC", "free": "0.01", "locked": "0"},
|
|
{"asset": "ETH", "free": "0.5", "locked": "0"},
|
|
{"asset": "DOGE", "free": "1", "locked": "0"},
|
|
]
|
|
}
|
|
|
|
def ticker_price(self, symbols=None):
|
|
mapping = {
|
|
"BTCUSDT": {"symbol": "BTCUSDT", "price": "60000"},
|
|
"ETHUSDT": {"symbol": "ETHUSDT", "price": "3000"},
|
|
"DOGEUSDT": {"symbol": "DOGEUSDT", "price": "0.1"},
|
|
}
|
|
return [mapping[symbol] for symbol in symbols]
|
|
|
|
def ticker_24h(self, symbols=None):
|
|
rows = {
|
|
"BTCUSDT": {"symbol": "BTCUSDT", "lastPrice": "60000", "priceChangePercent": "5", "quoteVolume": "9000000", "highPrice": "60200", "lowPrice": "55000"},
|
|
"ETHUSDT": {"symbol": "ETHUSDT", "lastPrice": "3000", "priceChangePercent": "3", "quoteVolume": "8000000", "highPrice": "3100", "lowPrice": "2800"},
|
|
"SOLUSDT": {"symbol": "SOLUSDT", "lastPrice": "150", "priceChangePercent": "8", "quoteVolume": "10000000", "highPrice": "152", "lowPrice": "130"},
|
|
"DOGEUSDT": {"symbol": "DOGEUSDT", "lastPrice": "0.1", "priceChangePercent": "1", "quoteVolume": "100", "highPrice": "0.11", "lowPrice": "0.09"},
|
|
}
|
|
if not symbols:
|
|
return list(rows.values())
|
|
return [rows[symbol] for symbol in symbols]
|
|
|
|
def exchange_info(self):
|
|
return {"symbols": [{"symbol": "BTCUSDT", "status": "TRADING"}, {"symbol": "ETHUSDT", "status": "TRADING"}, {"symbol": "SOLUSDT", "status": "TRADING"}, {"symbol": "DOGEUSDT", "status": "TRADING"}]}
|
|
|
|
def klines(self, symbol, interval, limit):
|
|
curves = {
|
|
"BTCUSDT": [50000, 52000, 54000, 56000, 58000, 59000, 60000],
|
|
"ETHUSDT": [2600, 2650, 2700, 2800, 2900, 2950, 3000],
|
|
"SOLUSDT": [120, 125, 130, 135, 140, 145, 150],
|
|
"DOGEUSDT": [0.11, 0.108, 0.105, 0.103, 0.102, 0.101, 0.1],
|
|
}[symbol]
|
|
rows = []
|
|
for index, close in enumerate(curves[-limit:]):
|
|
rows.append([index, close * 0.98, close * 1.01, close * 0.97, close, 100 + index * 10, index + 1, close * (100 + index * 10)])
|
|
return rows
|
|
|
|
|
|
class OpportunityServiceTestCase(unittest.TestCase):
|
|
def setUp(self):
|
|
self.config = {
|
|
"market": {"default_quote": "USDT", "universe_allowlist": [], "universe_denylist": []},
|
|
"trading": {"dust_usdt_threshold": 10.0},
|
|
"opportunity": {
|
|
"scan_limit": 10,
|
|
"top_n": 5,
|
|
"min_quote_volume": 1000.0,
|
|
"weights": {
|
|
"trend": 1.0,
|
|
"momentum": 1.0,
|
|
"breakout": 0.8,
|
|
"volume": 0.7,
|
|
"volatility_penalty": 0.5,
|
|
"position_concentration_penalty": 0.6,
|
|
},
|
|
},
|
|
}
|
|
|
|
def test_portfolio_analysis_ignores_dust_and_emits_recommendations(self):
|
|
events = []
|
|
with patch.object(opportunity_service, "audit_event", side_effect=lambda event, payload: events.append(event)):
|
|
payload = opportunity_service.analyze_portfolio(self.config, spot_client=FakeSpotClient())
|
|
symbols = [item["symbol"] for item in payload["recommendations"]]
|
|
self.assertNotIn("DOGEUSDT", symbols)
|
|
self.assertEqual(symbols, ["BTCUSDT", "ETHUSDT"])
|
|
self.assertEqual(events, ["opportunity_portfolio_generated"])
|
|
|
|
def test_scan_is_deterministic(self):
|
|
with patch.object(opportunity_service, "audit_event", return_value=None):
|
|
payload = opportunity_service.scan_opportunities(self.config | {"opportunity": self.config["opportunity"] | {"top_n": 2}}, spot_client=FakeSpotClient())
|
|
self.assertEqual([item["symbol"] for item in payload["recommendations"]], ["SOLUSDT", "BTCUSDT"])
|
|
|
|
def test_score_candidate_handles_empty_klines(self):
|
|
score, metrics = opportunity_service._score_candidate([], [], {"price_change_pct": 1.0}, {}, 0.0)
|
|
self.assertEqual(score, 0.0)
|
|
self.assertEqual(metrics["trend"], 0.0)
|