Fix concurrent API credit exhaustion — shared prefetch + dedup warnings
- telegram_bot: one ForexDataProvider shared across all SCAN_TIMEFRAMES; combined prefetch of all unique TFs up front so the 4 concurrent scans hit the warm cache instead of firing simultaneous API calls - data_provider: skip prefetch_batch for timeframes already in cache - data_provider: deduplicate "no data" warning per symbol (fires once, not once per TF) - data_provider: [live] info log on successful fetch; outlier-removal downgraded to debug Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
+14
-4
@@ -20,6 +20,7 @@ import requests
|
|||||||
from config import PlatformConfig, Timeframe, SYMBOL_DISPLAY, PIP_VALUES
|
from config import PlatformConfig, Timeframe, SYMBOL_DISPLAY, PIP_VALUES
|
||||||
|
|
||||||
logger = logging.getLogger(__name__)
|
logger = logging.getLogger(__name__)
|
||||||
|
_no_data_warned: set = set() # suppress duplicate "no data" warnings per symbol
|
||||||
|
|
||||||
|
|
||||||
class DataIntegrityError(Exception):
|
class DataIntegrityError(Exception):
|
||||||
@@ -151,7 +152,17 @@ class ForexDataProvider:
|
|||||||
if lookback_days is None:
|
if lookback_days is None:
|
||||||
lookback_days = self.config.data_lookback_days
|
lookback_days = self.config.data_lookback_days
|
||||||
|
|
||||||
|
now = time.time()
|
||||||
for tf in timeframes:
|
for tf in timeframes:
|
||||||
|
cached = all(
|
||||||
|
f"{s}_{tf.value}" in self._cache_timestamps
|
||||||
|
and (now - self._cache_timestamps[f"{s}_{tf.value}"]) < self.config.refresh_interval
|
||||||
|
and not self._cache.get(f"{s}_{tf.value}", pd.DataFrame()).empty
|
||||||
|
for s in symbols
|
||||||
|
)
|
||||||
|
if cached:
|
||||||
|
logger.debug(f"Skipping prefetch for {tf.value} — all {len(symbols)} symbols cached")
|
||||||
|
continue
|
||||||
try:
|
try:
|
||||||
self._batch_fetch_timeframe(symbols, tf, lookback_days)
|
self._batch_fetch_timeframe(symbols, tf, lookback_days)
|
||||||
time.sleep(1.0) # brief pause between TF requests
|
time.sleep(1.0) # brief pause between TF requests
|
||||||
@@ -332,10 +343,9 @@ class ForexDataProvider:
|
|||||||
|
|
||||||
values = sym_data.get("values")
|
values = sym_data.get("values")
|
||||||
if not values:
|
if not values:
|
||||||
logger.warning(
|
if td_sym not in _no_data_warned:
|
||||||
f"No data returned by Twelve Data for {td_sym} "
|
logger.warning(f"No data from Twelve Data for {td_sym} — symbol may require a paid plan.")
|
||||||
f"— symbol may require a paid plan."
|
_no_data_warned.add(td_sym)
|
||||||
)
|
|
||||||
continue
|
continue
|
||||||
|
|
||||||
records = [
|
records = [
|
||||||
|
|||||||
@@ -104,9 +104,9 @@ class SignalAggregator:
|
|||||||
fundamental analysis, and produces final graded signals.
|
fundamental analysis, and produces final graded signals.
|
||||||
"""
|
"""
|
||||||
|
|
||||||
def __init__(self, config: PlatformConfig):
|
def __init__(self, config: PlatformConfig, data_provider: Optional[ForexDataProvider] = None):
|
||||||
self.config = config
|
self.config = config
|
||||||
self.data_provider = ForexDataProvider(config)
|
self.data_provider = data_provider or ForexDataProvider(config)
|
||||||
self.tech_engine = TechnicalEngine(config.technical)
|
self.tech_engine = TechnicalEngine(config.technical)
|
||||||
self.news_engine = NewsEngine(config.news)
|
self.news_engine = NewsEngine(config.news)
|
||||||
|
|
||||||
|
|||||||
+14
-4
@@ -33,6 +33,7 @@ from config import (
|
|||||||
MAJOR_PAIRS, CROSS_PAIRS, SYMBOL_DISPLAY,
|
MAJOR_PAIRS, CROSS_PAIRS, SYMBOL_DISPLAY,
|
||||||
PlatformConfig, SignalStrength, Timeframe,
|
PlatformConfig, SignalStrength, Timeframe,
|
||||||
)
|
)
|
||||||
|
from data_provider import ForexDataProvider
|
||||||
from signal_aggregator import SignalAggregator, SignalReport
|
from signal_aggregator import SignalAggregator, SignalReport
|
||||||
|
|
||||||
logger = logging.getLogger(__name__)
|
logger = logging.getLogger(__name__)
|
||||||
@@ -282,16 +283,25 @@ class ForexTelegramBot:
|
|||||||
cfg.watch_pairs = list(pairs)
|
cfg.watch_pairs = list(pairs)
|
||||||
return cfg
|
return cfg
|
||||||
|
|
||||||
def _run_scan(self, tf: str, pairs: List[str]) -> List[SignalReport]:
|
def _run_scan(self, tf: str, pairs: List[str], provider: ForexDataProvider) -> List[SignalReport]:
|
||||||
return SignalAggregator(self._build_config(tf, pairs)).analyze_all_pairs()
|
return SignalAggregator(self._build_config(tf, pairs), data_provider=provider).analyze_all_pairs()
|
||||||
|
|
||||||
async def _analyze_all_tfs(
|
async def _analyze_all_tfs(
|
||||||
self, pairs: List[str]
|
self, pairs: List[str]
|
||||||
) -> Dict[str, List[SignalReport]]:
|
) -> Dict[str, List[SignalReport]]:
|
||||||
"""Run all SCAN_TIMEFRAMES concurrently, return {tf: [reports]}."""
|
"""Single combined prefetch, then run all SCAN_TIMEFRAMES concurrently off the warm cache."""
|
||||||
|
# Collect all unique timeframes needed across every TF scan
|
||||||
|
all_tfs = list({
|
||||||
|
tf
|
||||||
|
for tf_name in SCAN_TIMEFRAMES
|
||||||
|
for tf in [_TF_MAP[tf_name]] + _TF_CONFIRMATIONS.get(tf_name, [])
|
||||||
|
})
|
||||||
|
provider = ForexDataProvider(PlatformConfig())
|
||||||
|
provider.prefetch_batch(list(pairs), all_tfs)
|
||||||
|
|
||||||
loop = asyncio.get_running_loop()
|
loop = asyncio.get_running_loop()
|
||||||
tasks = {
|
tasks = {
|
||||||
tf: loop.run_in_executor(self._executor, self._run_scan, tf, list(pairs))
|
tf: loop.run_in_executor(self._executor, self._run_scan, tf, list(pairs), provider)
|
||||||
for tf in SCAN_TIMEFRAMES
|
for tf in SCAN_TIMEFRAMES
|
||||||
}
|
}
|
||||||
results: Dict[str, List[SignalReport]] = {}
|
results: Dict[str, List[SignalReport]] = {}
|
||||||
|
|||||||
Reference in New Issue
Block a user