Make yf.download() reentrant by removing shared module globals
`download()` used module-level dicts in `yfinance.shared` (`_DFS`, `_ERRORS`, `_TRACEBACKS`, `_ISINS`) as scratch space. Concurrent calls trampled each other; the previous fix wrapped the whole call in `shared._LOCK`, which silently serialized parallel downloads — hiding the race instead of removing it, and defeating the parallelism users reach for `download()` to get. - Introduce `_DownloadCtx` per call holding `dfs/errors/tracebacks/isins` plus a fine-grained lock for thread workers. - `_download_one`, `_download_one_threaded`, `_realign_dfs` take the ctx explicitly and write into it. - Drop the call-wide `shared._LOCK.acquire()`; concurrent `download()` calls now actually run in parallel. The polling loop on the threaded path reads `len(ctx.dfs)` under the ctx lock. - `PriceHistory` records soft errors (delisted, missing tz, repair failure) in `self._last_error` instead of `shared._ERRORS`; the download loop reads it after `Ticker.history()` returns. - `yfinance.shared` is left in place for import-compat with code that may still import it, but yfinance no longer reads or writes it. - Update existing test_multi to the new `_download_one(ctx, ...)` signature and add regression tests covering result separation, no-raise on concurrent calls, and a sentinel proving `shared._DFS` is no longer used as scratch space.
This commit is contained in:
Alessandro Colace
@@ -0,0 +1,57 @@
|
||||
"""Concurrent ``yf.download()`` calls must not share scratch state.
|
||||
|
||||
These tests assert that two concurrent ``download()`` calls each receive a
|
||||
DataFrame containing exactly the tickers they requested — no contamination,
|
||||
no exceptions — and that ``download()`` no longer relies on module-level
|
||||
dicts in ``yfinance.shared`` as scratch space.
|
||||
"""
|
||||
import unittest
|
||||
from concurrent.futures import ThreadPoolExecutor
|
||||
|
||||
from tests.context import yfinance as yf
|
||||
|
||||
|
||||
class TestDownloadConcurrency(unittest.TestCase):
|
||||
|
||||
def _fetch(self, tickers):
|
||||
return yf.download(
|
||||
tickers, period="5d", interval="1d",
|
||||
threads=False, progress=False, auto_adjust=False,
|
||||
)
|
||||
|
||||
def test_concurrent_downloads_keep_results_separate(self):
|
||||
chunk_a = ["AAPL", "MSFT", "GOOG"]
|
||||
chunk_b = ["NVDA", "META", "AMZN"]
|
||||
|
||||
with ThreadPoolExecutor(max_workers=2) as ex:
|
||||
futures = [ex.submit(self._fetch, chunk_a), ex.submit(self._fetch, chunk_b)]
|
||||
results = [f.result() for f in futures]
|
||||
|
||||
# Each result must contain exactly its requested tickers — neither
|
||||
# missing any nor leaking the other call's tickers.
|
||||
got_a = set(results[0].columns.get_level_values("Ticker"))
|
||||
got_b = set(results[1].columns.get_level_values("Ticker"))
|
||||
self.assertEqual(got_a, set(chunk_a))
|
||||
self.assertEqual(got_b, set(chunk_b))
|
||||
|
||||
def test_concurrent_downloads_do_not_raise(self):
|
||||
# Repeat several times to make races more likely to surface.
|
||||
chunks = [["AAPL", "MSFT"], ["NVDA", "META"], ["GOOG", "AMZN"], ["TSLA", "JPM"]]
|
||||
with ThreadPoolExecutor(max_workers=4) as ex:
|
||||
list(ex.map(self._fetch, chunks))
|
||||
|
||||
def test_download_does_not_use_module_globals(self):
|
||||
# Pre-populate shared._DFS with a sentinel; if download() still
|
||||
# depended on it as scratch space, the sentinel would either be
|
||||
# wiped or leak into the result.
|
||||
from yfinance import shared
|
||||
sentinel = object()
|
||||
shared._DFS = {"__SENTINEL__": sentinel}
|
||||
|
||||
df = self._fetch(["AAPL"])
|
||||
self.assertEqual(set(df.columns.get_level_values("Ticker")), {"AAPL"})
|
||||
self.assertIs(shared._DFS.get("__SENTINEL__"), sentinel)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
+5
-4
@@ -6,7 +6,6 @@ from unittest.mock import patch
|
||||
import pandas as pd
|
||||
|
||||
from tests.context import yfinance as yf
|
||||
from yfinance import shared
|
||||
|
||||
|
||||
class TestDownloadThreadSafety(unittest.TestCase):
|
||||
@@ -23,10 +22,12 @@ class TestDownloadThreadSafety(unittest.TestCase):
|
||||
index=idx,
|
||||
)
|
||||
|
||||
def mock_download_one(ticker, *args, **kwargs):
|
||||
def mock_download_one(ctx, ticker, *args, **kwargs):
|
||||
time.sleep(0.05)
|
||||
df = aapl_df if ticker.upper() == 'AAPL' else msft_df
|
||||
shared._DFS[ticker.upper()] = df
|
||||
sym = ticker.upper()
|
||||
df = aapl_df if sym == 'AAPL' else msft_df
|
||||
with ctx.lock:
|
||||
ctx.dfs[sym] = df
|
||||
return df
|
||||
|
||||
results = {}
|
||||
|
||||
+95
-107
@@ -22,6 +22,7 @@
|
||||
from __future__ import print_function
|
||||
|
||||
import logging
|
||||
import threading
|
||||
import time as _time
|
||||
import traceback
|
||||
from typing import Union
|
||||
@@ -32,10 +33,23 @@ from curl_cffi import requests
|
||||
|
||||
from . import Ticker, utils
|
||||
from .data import YfData
|
||||
from . import shared
|
||||
from .config import YfConfig
|
||||
from .const import period_default
|
||||
|
||||
|
||||
class _DownloadCtx:
|
||||
"""Per-call scratch state for download(). Concurrent calls each get
|
||||
their own instance, so no shared mutation between threads."""
|
||||
__slots__ = ('dfs', 'errors', 'tracebacks', 'isins', 'progress_bar', 'lock')
|
||||
|
||||
def __init__(self):
|
||||
self.dfs = {}
|
||||
self.errors = {}
|
||||
self.tracebacks = {}
|
||||
self.isins = {}
|
||||
self.progress_bar = None
|
||||
self.lock = threading.Lock()
|
||||
|
||||
@utils.log_indent_decorator
|
||||
def download(tickers, start=None, end=None, actions=False, threads=True,
|
||||
ignore_tz=None, group_by='column', auto_adjust=True, back_adjust=False,
|
||||
@@ -92,20 +106,17 @@ def download(tickers, start=None, end=None, actions=False, threads=True,
|
||||
multi_level_index: bool
|
||||
Optional. Always return a MultiIndex DataFrame? Default is True
|
||||
"""
|
||||
shared._LOCK.acquire()
|
||||
try:
|
||||
return _download_impl(
|
||||
tickers, start=start, end=end, actions=actions, threads=threads,
|
||||
ignore_tz=ignore_tz, group_by=group_by, auto_adjust=auto_adjust,
|
||||
back_adjust=back_adjust, repair=repair, keepna=keepna, progress=progress,
|
||||
period=period, interval=interval, prepost=prepost, rounding=rounding,
|
||||
timeout=timeout, session=session, multi_level_index=multi_level_index,
|
||||
)
|
||||
finally:
|
||||
shared._LOCK.release()
|
||||
return _download_impl(
|
||||
_DownloadCtx(),
|
||||
tickers, start=start, end=end, actions=actions, threads=threads,
|
||||
ignore_tz=ignore_tz, group_by=group_by, auto_adjust=auto_adjust,
|
||||
back_adjust=back_adjust, repair=repair, keepna=keepna, progress=progress,
|
||||
period=period, interval=interval, prepost=prepost, rounding=rounding,
|
||||
timeout=timeout, session=session, multi_level_index=multi_level_index,
|
||||
)
|
||||
|
||||
|
||||
def _download_impl(tickers, start=None, end=None, actions=False, threads=True,
|
||||
def _download_impl(ctx, tickers, start=None, end=None, actions=False, threads=True,
|
||||
ignore_tz=None, group_by='column', auto_adjust=True, back_adjust=False,
|
||||
repair=False, keepna=False, progress=True, period=period_default, interval="1d",
|
||||
prepost=False, rounding=False, timeout=10, session=None,
|
||||
@@ -113,127 +124,96 @@ def _download_impl(tickers, start=None, end=None, actions=False, threads=True,
|
||||
logger = utils.get_yf_logger()
|
||||
session = session or requests.Session(impersonate="chrome")
|
||||
|
||||
# Ensure data initialised with session.
|
||||
YfData(session=session)
|
||||
|
||||
if logger.isEnabledFor(logging.DEBUG):
|
||||
if threads:
|
||||
# With DEBUG, each thread generates a lot of log messages.
|
||||
# And with multi-threading, these messages will be interleaved, bad!
|
||||
# So disable multi-threading to make log readable.
|
||||
# multi-threaded log messages would interleave; serialize.
|
||||
logger.debug('Disabling multithreading because DEBUG logging enabled')
|
||||
threads = False
|
||||
if progress:
|
||||
# Disable progress bar, interferes with display of log messages
|
||||
progress = False
|
||||
|
||||
if ignore_tz is None:
|
||||
# Set default value depending on interval
|
||||
if interval[-1] in ['m', 'h']:
|
||||
# Intraday
|
||||
ignore_tz = False
|
||||
else:
|
||||
ignore_tz = True
|
||||
ignore_tz = interval[-1] not in ('m', 'h')
|
||||
|
||||
# create ticker list
|
||||
tickers = tickers if isinstance(
|
||||
tickers, (list, set, tuple)) else tickers.replace(',', ' ').split()
|
||||
|
||||
# accept isin as ticker
|
||||
shared._ISINS = {}
|
||||
_tickers_ = []
|
||||
for ticker in tickers:
|
||||
if utils.is_isin(ticker):
|
||||
isin = ticker
|
||||
ticker = utils.get_ticker_by_isin(ticker)
|
||||
shared._ISINS[ticker] = isin
|
||||
ctx.isins[ticker] = isin
|
||||
_tickers_.append(ticker)
|
||||
|
||||
tickers = _tickers_
|
||||
|
||||
tickers = list(set([ticker.upper() for ticker in tickers]))
|
||||
tickers = list(set([t.upper() for t in _tickers_]))
|
||||
|
||||
if progress:
|
||||
shared._PROGRESS_BAR = utils.ProgressBar(len(tickers), 'completed')
|
||||
ctx.progress_bar = utils.ProgressBar(len(tickers), 'completed')
|
||||
|
||||
# reset shared._DFS
|
||||
shared._DFS = {}
|
||||
shared._ERRORS = {}
|
||||
shared._TRACEBACKS = {}
|
||||
|
||||
# download using threads
|
||||
if threads:
|
||||
if threads is True:
|
||||
threads = min([len(tickers), _multitasking.cpu_count() * 2])
|
||||
_multitasking.set_max_threads(threads)
|
||||
for i, ticker in enumerate(tickers):
|
||||
_download_one_threaded(ticker, period=period, interval=interval,
|
||||
_download_one_threaded(ctx, ticker, period=period, interval=interval,
|
||||
start=start, end=end, prepost=prepost,
|
||||
actions=actions, auto_adjust=auto_adjust,
|
||||
back_adjust=back_adjust, repair=repair, keepna=keepna,
|
||||
progress=(progress and i > 0),
|
||||
rounding=rounding, timeout=timeout)
|
||||
while len(shared._DFS) < len(tickers):
|
||||
while True:
|
||||
with ctx.lock:
|
||||
if len(ctx.dfs) >= len(tickers):
|
||||
break
|
||||
_time.sleep(0.01)
|
||||
# download synchronously
|
||||
else:
|
||||
for i, ticker in enumerate(tickers):
|
||||
data = _download_one(ticker, period=period, interval=interval,
|
||||
start=start, end=end, prepost=prepost,
|
||||
actions=actions, auto_adjust=auto_adjust,
|
||||
back_adjust=back_adjust, repair=repair, keepna=keepna,
|
||||
rounding=rounding, timeout=timeout)
|
||||
_download_one(ctx, ticker, period=period, interval=interval,
|
||||
start=start, end=end, prepost=prepost,
|
||||
actions=actions, auto_adjust=auto_adjust,
|
||||
back_adjust=back_adjust, repair=repair, keepna=keepna,
|
||||
rounding=rounding, timeout=timeout)
|
||||
if progress:
|
||||
shared._PROGRESS_BAR.animate()
|
||||
ctx.progress_bar.animate()
|
||||
|
||||
if progress:
|
||||
shared._PROGRESS_BAR.completed()
|
||||
ctx.progress_bar.completed()
|
||||
|
||||
if shared._ERRORS:
|
||||
# Send errors to logging module
|
||||
logger = utils.get_yf_logger()
|
||||
if ctx.errors:
|
||||
logger.error('\n%.f Failed download%s:' % (
|
||||
len(shared._ERRORS), 's' if len(shared._ERRORS) > 1 else ''))
|
||||
len(ctx.errors), 's' if len(ctx.errors) > 1 else ''))
|
||||
|
||||
# Log each distinct error once, with list of symbols affected
|
||||
errors = {}
|
||||
for ticker in shared._ERRORS:
|
||||
err = shared._ERRORS[ticker]
|
||||
for ticker, err in ctx.errors.items():
|
||||
err = err.replace(f'${ticker}: ', '')
|
||||
if err not in errors:
|
||||
errors[err] = [ticker]
|
||||
else:
|
||||
errors[err].append(ticker)
|
||||
for err in errors.keys():
|
||||
logger.error(f'{errors[err]}: ' + err)
|
||||
errors.setdefault(err, []).append(ticker)
|
||||
for err, syms in errors.items():
|
||||
logger.error(f'{syms}: ' + err)
|
||||
|
||||
# Log each distinct traceback once, with list of symbols affected
|
||||
tbs = {}
|
||||
for ticker in shared._TRACEBACKS:
|
||||
tb = shared._TRACEBACKS[ticker]
|
||||
for ticker, tb in ctx.tracebacks.items():
|
||||
tb = tb.replace(f'${ticker}: ', '')
|
||||
if tb not in tbs:
|
||||
tbs[tb] = [ticker]
|
||||
else:
|
||||
tbs[tb].append(ticker)
|
||||
for tb in tbs.keys():
|
||||
logger.debug(f'{tbs[tb]}: ' + tb)
|
||||
tbs.setdefault(tb, []).append(ticker)
|
||||
for tb, syms in tbs.items():
|
||||
logger.debug(f'{syms}: ' + tb)
|
||||
|
||||
if ignore_tz:
|
||||
for tkr in shared._DFS.keys():
|
||||
if (shared._DFS[tkr] is not None) and (shared._DFS[tkr].shape[0] > 0):
|
||||
shared._DFS[tkr].index = shared._DFS[tkr].index.tz_localize(None)
|
||||
for tkr, df in ctx.dfs.items():
|
||||
if df is not None and df.shape[0] > 0:
|
||||
df.index = df.index.tz_localize(None)
|
||||
|
||||
try:
|
||||
data = _pd.concat(shared._DFS.values(), axis=1, sort=True,
|
||||
keys=shared._DFS.keys(), names=['Ticker', 'Price'])
|
||||
data = _pd.concat(ctx.dfs.values(), axis=1, sort=True,
|
||||
keys=ctx.dfs.keys(), names=['Ticker', 'Price'])
|
||||
except Exception:
|
||||
_realign_dfs()
|
||||
data = _pd.concat(shared._DFS.values(), axis=1, sort=True,
|
||||
keys=shared._DFS.keys(), names=['Ticker', 'Price'])
|
||||
_realign_dfs(ctx)
|
||||
data = _pd.concat(ctx.dfs.values(), axis=1, sort=True,
|
||||
keys=ctx.dfs.keys(), names=['Ticker', 'Price'])
|
||||
data.index = _pd.to_datetime(data.index, utc=not ignore_tz)
|
||||
# switch names back to isins if applicable
|
||||
data.rename(columns=shared._ISINS, inplace=True)
|
||||
data.rename(columns=ctx.isins, inplace=True)
|
||||
|
||||
if group_by == 'column':
|
||||
data.columns = data.columns.swaplevel(0, 1)
|
||||
@@ -245,64 +225,72 @@ def _download_impl(tickers, start=None, end=None, actions=False, threads=True,
|
||||
return data
|
||||
|
||||
|
||||
def _realign_dfs():
|
||||
def _realign_dfs(ctx):
|
||||
idx_len = 0
|
||||
idx = None
|
||||
|
||||
for df in shared._DFS.values():
|
||||
for df in ctx.dfs.values():
|
||||
if len(df) > idx_len:
|
||||
idx_len = len(df)
|
||||
idx = df.index
|
||||
|
||||
for key in shared._DFS.keys():
|
||||
for key in list(ctx.dfs.keys()):
|
||||
try:
|
||||
shared._DFS[key] = _pd.DataFrame(
|
||||
index=idx, data=shared._DFS[key]).drop_duplicates()
|
||||
ctx.dfs[key] = _pd.DataFrame(
|
||||
index=idx, data=ctx.dfs[key]).drop_duplicates()
|
||||
except Exception:
|
||||
shared._DFS[key] = _pd.concat([
|
||||
utils.empty_df(idx), shared._DFS[key].dropna()
|
||||
ctx.dfs[key] = _pd.concat([
|
||||
utils.empty_df(idx), ctx.dfs[key].dropna()
|
||||
], axis=0, sort=True)
|
||||
|
||||
# remove duplicate index
|
||||
shared._DFS[key] = shared._DFS[key].loc[
|
||||
~shared._DFS[key].index.duplicated(keep='last')]
|
||||
ctx.dfs[key] = ctx.dfs[key].loc[
|
||||
~ctx.dfs[key].index.duplicated(keep='last')]
|
||||
|
||||
|
||||
@_multitasking.task
|
||||
def _download_one_threaded(ticker, start=None, end=None,
|
||||
def _download_one_threaded(ctx, ticker, start=None, end=None,
|
||||
auto_adjust=False, back_adjust=False, repair=False,
|
||||
actions=False, progress=True, period=None,
|
||||
interval="1d", prepost=False,
|
||||
keepna=False, rounding=False, timeout=10):
|
||||
_download_one(ticker, start, end, auto_adjust, back_adjust, repair,
|
||||
actions, period, interval, prepost, rounding,
|
||||
keepna, timeout)
|
||||
_download_one(ctx, ticker, start, end, auto_adjust, back_adjust, repair,
|
||||
actions, period, interval, prepost, rounding,
|
||||
keepna, timeout)
|
||||
if progress:
|
||||
shared._PROGRESS_BAR.animate()
|
||||
ctx.progress_bar.animate()
|
||||
|
||||
|
||||
def _download_one(ticker, start=None, end=None,
|
||||
def _download_one(ctx, ticker, start=None, end=None,
|
||||
auto_adjust=False, back_adjust=False, repair=False,
|
||||
actions=False, period=None, interval="1d",
|
||||
prepost=False, rounding=False,
|
||||
keepna=False, timeout=10):
|
||||
data = None
|
||||
|
||||
sym = ticker.upper()
|
||||
|
||||
backup = YfConfig.network.hide_exceptions
|
||||
YfConfig.network.hide_exceptions = False
|
||||
try:
|
||||
data = Ticker(ticker).history(
|
||||
period=period, interval=interval,
|
||||
start=start, end=end, prepost=prepost,
|
||||
actions=actions, auto_adjust=auto_adjust,
|
||||
back_adjust=back_adjust, repair=repair,
|
||||
rounding=rounding, keepna=keepna, timeout=timeout
|
||||
tkr = Ticker(ticker)
|
||||
data = tkr.history(
|
||||
period=period, interval=interval,
|
||||
start=start, end=end, prepost=prepost,
|
||||
actions=actions, auto_adjust=auto_adjust,
|
||||
back_adjust=back_adjust, repair=repair,
|
||||
rounding=rounding, keepna=keepna, timeout=timeout
|
||||
)
|
||||
shared._DFS[ticker.upper()] = data
|
||||
with ctx.lock:
|
||||
ctx.dfs[sym] = data
|
||||
# PriceHistory records soft errors (e.g. delisted, missing tz)
|
||||
# without raising; surface them so download() can log them.
|
||||
ph = tkr._price_history
|
||||
if ph is not None and ph._last_error is not None:
|
||||
ctx.errors[sym] = ph._last_error
|
||||
except Exception as e:
|
||||
shared._DFS[ticker.upper()] = utils.empty_df()
|
||||
shared._ERRORS[ticker.upper()] = repr(e)
|
||||
shared._TRACEBACKS[ticker.upper()] = traceback.format_exc()
|
||||
with ctx.lock:
|
||||
ctx.dfs[sym] = utils.empty_df()
|
||||
ctx.errors[sym] = repr(e)
|
||||
ctx.tracebacks[sym] = traceback.format_exc()
|
||||
|
||||
YfConfig.network.hide_exceptions = backup
|
||||
|
||||
|
||||
@@ -9,7 +9,7 @@ import pandas as pd
|
||||
import time as _time
|
||||
import warnings
|
||||
|
||||
from yfinance import shared, utils
|
||||
from yfinance import utils
|
||||
from yfinance.config import YfConfig
|
||||
from yfinance.const import _BASE_URL_, _PRICE_COLNAMES_, period_default, _SENTINEL_
|
||||
from yfinance.exceptions import YFDataException, YFInvalidPeriodError, YFPricesMissingError, YFRateLimitError, YFTzMissingError
|
||||
@@ -32,6 +32,8 @@ class PriceHistory:
|
||||
# Limit recursion depth when repairing prices
|
||||
self._reconstruct_start_interval = None
|
||||
|
||||
self._last_error = None
|
||||
|
||||
@utils.log_indent_decorator
|
||||
def history(self, period=period_default, interval="1d",
|
||||
start=None, end=None, prepost=False, actions=True,
|
||||
@@ -105,8 +107,7 @@ class PriceHistory:
|
||||
# Every valid ticker has a timezone. A missing timezone is a problem.
|
||||
_exception = YFTzMissingError(self.ticker)
|
||||
err_msg = str(_exception)
|
||||
shared._DFS[self.ticker] = utils.empty_df()
|
||||
shared._ERRORS[self.ticker] = err_msg.split(': ', 1)[1]
|
||||
self._last_error = err_msg.split(': ', 1)[1]
|
||||
if raise_errors or (not YfConfig.debug.hide_exceptions):
|
||||
raise _exception
|
||||
else:
|
||||
@@ -131,8 +132,7 @@ class PriceHistory:
|
||||
# Every valid ticker has a timezone. A missing timezone is a problem.
|
||||
_exception = YFTzMissingError(self.ticker)
|
||||
err_msg = str(_exception)
|
||||
shared._DFS[self.ticker] = utils.empty_df()
|
||||
shared._ERRORS[self.ticker] = err_msg.split(': ', 1)[1]
|
||||
self._last_error = err_msg.split(': ', 1)[1]
|
||||
if raise_errors or (not YfConfig.debug.hide_exceptions):
|
||||
raise _exception
|
||||
else:
|
||||
@@ -289,8 +289,7 @@ class PriceHistory:
|
||||
|
||||
if fail:
|
||||
err_msg = str(_exception)
|
||||
shared._DFS[self.ticker] = utils.empty_df()
|
||||
shared._ERRORS[self.ticker] = err_msg.split(': ', 1)[1]
|
||||
self._last_error = err_msg.split(': ', 1)[1]
|
||||
if raise_errors or (not YfConfig.debug.hide_exceptions):
|
||||
raise _exception
|
||||
else:
|
||||
@@ -504,8 +503,7 @@ class PriceHistory:
|
||||
err_msg = "auto_adjust failed with %s" % e
|
||||
else:
|
||||
err_msg = "back_adjust failed with %s" % e
|
||||
shared._DFS[self.ticker] = utils.empty_df()
|
||||
shared._ERRORS[self.ticker] = err_msg
|
||||
self._last_error = err_msg
|
||||
logger.error('%s: %s' % (self.ticker, err_msg))
|
||||
|
||||
if rounding:
|
||||
|
||||
Reference in New Issue
Block a user