mirror of
https://github.com/OpenByteInc/QuantDinger.git
synced 2026-09-28 15:22:56 +08:00
Claude Code
eee8b02af2
Adds a read-only MOEX data source backed by the public MOEX ISS HTTP API (https://iss.moex.com/iss). Targets the TQBR equities board so common Russian tickers (SBER, GAZP, LKOH, ...) work out of the box. What's new - app/data_sources/moex.py: MOEXDataSource with get_kline / get_ticker - QuantDinger timeframes mapped to ISS intervals; 5m/15m/30m/4H are resampled from the nearest finer native interval (1m or 60m) - ISS naive timestamps treated as Europe/Moscow (UTC+3) and converted to Unix UTC seconds - Pagination via ISS start parameter - Symbol normalization handles SBER, SBER.ME, GAZP:MOEX, etc. - app/data_sources/factory.py: registers MOEX market + aliases - app/routes/market.py: MOEX in /market/types between HKStock and Crypto - app/services/symbol_name.py: resolves MOEX names from ISS securities endpoint - app/services/strategy.py: explicit guard - MOEX cannot be used as a live trading market_category (create / batch / update paths) - tests/test_moex_data_source.py: 11 offline unit tests (mocked HTTP) - scripts/verify_moex.py: live ISS smoke-test script Limitations - Analysis & backtesting only. No live order placement on MOEX. - TQBR board only by default. Other boards can be enabled by constructing MOEXDataSource(board=...) explicitly. - Sub-hour non-native timeframes (5m/15m/30m) and 4H are resampled client-side; large limits at fine timeframes pull more data.
57 lines
1.9 KiB
Python
Executable File
57 lines
1.9 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""Live verification script for the MOEX data source.
|
|
|
|
Hits the public MOEX ISS API. Run from the backend_api_python directory:
|
|
|
|
python scripts/verify_moex.py
|
|
python scripts/verify_moex.py SBER 1D 30
|
|
python scripts/verify_moex.py GAZP 1H 50
|
|
|
|
Prints a summary of fetched candles and the latest ticker. Useful for smoke-
|
|
testing connectivity in environments where pytest can run only offline tests.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import sys
|
|
from datetime import datetime, timezone
|
|
|
|
# Make `app` importable when the script is run directly
|
|
sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
|
|
|
|
# Minimal env so app.config doesn't blow up
|
|
os.environ.setdefault("SECRET_KEY", "verify-moex-script")
|
|
os.environ.setdefault("ADMIN_USER", "verify")
|
|
os.environ.setdefault("ADMIN_PASSWORD", "verifypass")
|
|
|
|
from app.data_sources.factory import DataSourceFactory # noqa: E402
|
|
|
|
|
|
def main() -> int:
|
|
symbol = sys.argv[1] if len(sys.argv) > 1 else "SBER"
|
|
timeframe = sys.argv[2] if len(sys.argv) > 2 else "1D"
|
|
limit = int(sys.argv[3]) if len(sys.argv) > 3 else 10
|
|
|
|
print(f"== MOEX verification == symbol={symbol} timeframe={timeframe} limit={limit}")
|
|
src = DataSourceFactory.get_source("MOEX")
|
|
print(f"data source: {src.name}")
|
|
|
|
klines = src.get_kline(symbol, timeframe, limit)
|
|
print(f"candles returned: {len(klines)}")
|
|
for k in klines[-5:]:
|
|
utc = datetime.fromtimestamp(k['time'], tz=timezone.utc).isoformat()
|
|
print(f" {utc} O={k['open']:>8} H={k['high']:>8} L={k['low']:>8} C={k['close']:>8} V={k['volume']}")
|
|
|
|
print()
|
|
ticker = src.get_ticker(symbol)
|
|
print(f"ticker: {ticker}")
|
|
|
|
if not klines:
|
|
print("WARNING: no candles returned (network blocked? bad symbol? non-trading window?)")
|
|
return 1
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|