156 lines
5.7 KiB
Python
156 lines
5.7 KiB
Python
"""Signal, opportunity, and portfolio service tests."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
from coinhunter.services import opportunity_service, portfolio_service, signal_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_stats(self, symbols=None, *, window="1d"):
|
|
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},
|
|
"signal": {
|
|
"lookback_interval": "1h",
|
|
"trend": 1.0,
|
|
"momentum": 1.0,
|
|
"breakout": 0.8,
|
|
"volume": 0.7,
|
|
"volatility_penalty": 0.5,
|
|
},
|
|
"opportunity": {
|
|
"scan_limit": 10,
|
|
"top_n": 5,
|
|
"min_quote_volume": 1000.0,
|
|
"entry_threshold": 1.5,
|
|
"watch_threshold": 0.6,
|
|
"overlap_penalty": 0.6,
|
|
},
|
|
"portfolio": {
|
|
"add_threshold": 1.5,
|
|
"hold_threshold": 0.6,
|
|
"trim_threshold": 0.2,
|
|
"exit_threshold": -0.2,
|
|
"max_position_weight": 0.6,
|
|
},
|
|
}
|
|
|
|
def test_portfolio_analysis_ignores_dust_and_emits_recommendations(self):
|
|
events = []
|
|
with patch.object(portfolio_service, "audit_event", side_effect=lambda event, payload, **kwargs: events.append(event)):
|
|
payload = portfolio_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(payload["recommendations"][0]["action"], "add")
|
|
self.assertEqual(payload["recommendations"][1]["action"], "hold")
|
|
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"])
|
|
self.assertEqual([item["action"] for item in payload["recommendations"]], ["enter", "enter"])
|
|
|
|
def test_signal_score_handles_empty_klines(self):
|
|
score, metrics = signal_service.score_market_signal([], [], {"price_change_pct": 1.0}, {})
|
|
self.assertEqual(score, 0.0)
|
|
self.assertEqual(metrics["trend"], 0.0)
|