harry/ForexBot2
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Remove WebSocket: use candle-boundary scanning only

Browse Source 
- Remove websockets import, _ws_monitor, _handle_breach, _breach_tasks
- Remove _WS_TO_PAIR, _WS_SYMBOLS, _WS_URL, _WS_COOLDOWN constants
- Remove _last_prices, _scan_cooldowns, _cleanup_cooldowns
- Remove ws_cooldown_seconds from PlatformConfig
- Remove websockets from requirements.txt
- Simplify _check_level_breach to proximity-based detection
- Update dashboard to use report prices instead of WS prices
- Clean up status/help messages, _start/_stop_scheduler
- Update tests to match new level breach behavior

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
harwinder
2026-05-18 17:38:25 -04:00
parent de57c016d1
commit 541ef7ff4a
4 changed files with 20 additions and 202 deletions
Download Patch File Download Diff File Expand all files Collapse all files
test_all.py
+10 -31
View File
@@ -1324,8 +1324,7 @@ from testing_telegram_bot import (
_seconds_until_boundary, _check_level_breach,
_format_stats_dashboard,
scan_keyboard, SCAN_BUTTONS, SCAN_LOOKUP,
_update_level_cache, _last_prices, _level_cache, _level_cache_ts,
_WS_TO_PAIR,
_update_level_cache, _level_cache, _level_cache_ts,
)
@@ -1428,65 +1427,50 @@ def _():
assert result <= 240 * 60 + 10 + 1
@test("_check_level_breach detects upward cross")
@test("_check_level_breach detects price near level")
def _():
_level_cache["TEST_PAIR"] = {"swing_high_1": 1.1000}
_level_cache_ts["TEST_PAIR"] = time.time()
_last_prices["TEST_PAIR"] = 1.0999
result = _check_level_breach("TEST_PAIR", 1.1001)
result = _check_level_breach("TEST_PAIR", 1.10005)
assert result == "swing_high_1"
# cleanup
del _level_cache["TEST_PAIR"]
del _level_cache_ts["TEST_PAIR"]
del _last_prices["TEST_PAIR"]
@test("_check_level_breach detects downward cross")
@test("_check_level_breach detects price at exact level")
def _():
_level_cache["TEST_PAIR"] = {"swing_low_1": 1.0900}
_level_cache_ts["TEST_PAIR"] = time.time()
_last_prices["TEST_PAIR"] = 1.0901
result = _check_level_breach("TEST_PAIR", 1.0899)
result = _check_level_breach("TEST_PAIR", 1.0900)
assert result == "swing_low_1"
del _level_cache["TEST_PAIR"]
del _level_cache_ts["TEST_PAIR"]
del _last_prices["TEST_PAIR"]
@test("_check_level_breach no breach when price stays same side")
@test("_check_level_breach no breach when price is far from level")
def _():
_level_cache["TEST_PAIR"] = {"swing_high_1": 1.1000}
_level_cache_ts["TEST_PAIR"] = time.time()
_last_prices["TEST_PAIR"] = 1.0990
result = _check_level_breach("TEST_PAIR", 1.0995)
result = _check_level_breach("TEST_PAIR", 1.0950)
assert result is None
del _level_cache["TEST_PAIR"]
del _level_cache_ts["TEST_PAIR"]
del _last_prices["TEST_PAIR"]
@test("_check_level_breach returns None for expired cache")
def _():
_level_cache["TEST_PAIR"] = {"swing_high_1": 1.1000}
_level_cache_ts["TEST_PAIR"] = time.time() - 7200 # 2h old
_last_prices["TEST_PAIR"] = 1.0999
result = _check_level_breach("TEST_PAIR", 1.1001)
result = _check_level_breach("TEST_PAIR", 1.1000)
assert result is None
del _level_cache["TEST_PAIR"]
del _level_cache_ts["TEST_PAIR"]
del _last_prices["TEST_PAIR"]
@test("_check_level_breach returns None when no previous price")
@test("_check_level_breach returns None for empty cache")
def _():
_level_cache["TEST_PAIR"] = {"swing_high_1": 1.1000}
_level_cache_ts["TEST_PAIR"] = time.time()
if "TEST_PAIR" in _last_prices:
del _last_prices["TEST_PAIR"]
result = _check_level_breach("TEST_PAIR", 1.1001)
result = _check_level_breach("NONEXISTENT_PAIR", 1.1000)
assert result is None
del _level_cache["TEST_PAIR"]
del _level_cache_ts["TEST_PAIR"]
@test("SCAN_BUTTONS and SCAN_LOOKUP consistency")
@@ -1524,11 +1508,6 @@ def _():
_level_cache_ts.clear()
@test("_WS_TO_PAIR mapping is correct")
def _():
assert _WS_TO_PAIR["XAU/USD"] == XAUUSD
@test("_format_stats_dashboard returns waiting message when empty")
def _():
import testing_telegram_bot