异步问题。 "Future pending"
ASYNCIO Issues. "Future pending"
我目前在使用 Joblib 运行 多处理或并行程序时遇到问题。我之前能够让它工作，并且我总共达到了 1 分钟的时间，但是，我四处走动并改变了很多，并且弄乱了一些东西。我已经发布了准系统代码，因为我收到了同样的错误。我正在尝试遍历所有 150 个股票代码，并使用 yahoo finance 来接收每个股票代码的期权链。我正在尝试在一分钟内完成此操作。我也尝试过其他库，比如 asyncio，但没有成功。任何建议将不胜感激。
import yfinance as yf
def background(f):
def wrapped(*args, **kwargs):
return asyncio.get_event_loop().run_in_executor(None, f, *args, **kwargs)
return wrapped
done = []
@background
def downloadChain(ticker):
print(ticker)
df = pd.DataFrame()
daysOut = 100
chain = 0
try:
yf_ticker = yf.Ticker(ticker)
expiration_dates = yf_ticker.options
for expiration_date in expiration_dates:
if (datetime.fromisoformat(expiration_date) - datetime.now()).days <= daysOut:
try:
chain = yf_ticker.option_chain(expiration_date)
df = df.append(chain)
except Exception as e:
pass
except Exception as e:
pass
done.append(ticker)
主要功能：
symbols = ["WATT","TSLA","UVXY","VXX","KEYS","EGO","GLD","WORK","BYND","BLK","PINS","LYFT","SPCE","PAYC","WDAY","UBER","CHGG","SHAK","CMG","CTL","ACB","TLRY","CGC","MJ","ORCL","GRUB","RNG","JWN","TTWO","ADI","ATVI","EA","SNE","GAMR","TXN","TMUS","MCHP","TSM","XBI","ETFC","MS","IWM","EXPD","RCL","CCL","MOMO","BABA","VMW","CRM","ULTA","SKYY","SPLK","FLWS","AVGO","TWTR","PANW","RJF","SABR","LOW","RS","ON","VEEV","DOCU","FB","SNAP","HPQ","RACE","F","AMAT","MRO","STM","AAL","DAL","VICR","XLC","CRON","DELL","T","VZ","S","MELI","CVM","REGN","NVAX","APT","CODX","LAKE","MRNA","EBS","INO", "SPY","SH","QQQ","XLF","KRE","XLV","HYG","LQD","NET","NFLX","ROKU","SHOP","AMZN","AAPL","MSFT","GOOGL","GOOG","NVDA","MU","AMD","INTC","MRVL","QCOMM","SQ","PYPL","TTD","TSLA","ZM","TDOC","LVGO","MDB","HD","VNQ","ARI","ACC","IIPR","EQR","EPR","SPG","PLD","ACB","WHR","NVAX","APT","MDT","CLRX","COST","SDC","LK","PVH","KSS","M","LULU","NKE","KO","BAC","JPM","CS","WFC","ARKW","ARKK","MGM","AMAT","WYNN","TGT","ITT","FXI"]
for ticker in symbols:
downloadChain(ticker)
我添加了一个单独的循环来查看 "done" 数组的大小，它包含所有已完成的符号。我不确定我做了什么更改，但现在这个循环将在大约 10-15 分钟内完成，而预计需要 1 分钟。
while True:
clear_output(wait=True)
print(len(done))
"fix"有两个版本。将它们添加为答案而不是将评论用作聊天 :)
import asyncio
import pandas as pd
import yfinance as yf
from concurrent.futures import ThreadPoolExecutor
def background(f):
def wrapped(*args, **kwargs):
return asyncio.get_event_loop().run_in_executor(executor, f, *args, **kwargs)
return wrapped
done = []
@background
def downloadChain(ticker):
print(ticker)
df = pd.DataFrame()
daysOut = 100
chain = 0
try:
yf_ticker = yf.Ticker(ticker)
expiration_dates = yf_ticker.options
for expiration_date in expiration_dates:
if (datetime.fromisoformat(expiration_date) - datetime.now()).days <= daysOut:
try:
chain = yf_ticker.option_chain(expiration_date)
df = df.append(chain)
except Exception as e:
pass
except Exception as e:
pass
done.append(ticker)
symbols = ["WATT","TSLA","UVXY","VXX","KEYS","EGO","GLD","WORK","BYND","BLK","PINS","LYFT","SPCE","PAYC","WDAY","UBER","CHGG","SHAK","CMG","CTL","ACB","TLRY","CGC","MJ","ORCL","GRUB","RNG","JWN","TTWO","ADI","ATVI","EA","SNE","GAMR","TXN","TMUS","MCHP","TSM","XBI","ETFC","MS","IWM","EXPD","RCL","CCL","MOMO","BABA","VMW","CRM","ULTA","SKYY","SPLK","FLWS","AVGO","TWTR","PANW","RJF","SABR","LOW","RS","ON","VEEV","DOCU","FB","SNAP","HPQ","RACE","F","AMAT","MRO","STM","AAL","DAL","VICR","XLC","CRON","DELL","T","VZ","S","MELI","CVM","REGN","NVAX","APT","CODX","LAKE","MRNA","EBS","INO", "SPY","SH","QQQ","XLF","KRE","XLV","HYG","LQD","NET","NFLX","ROKU","SHOP","AMZN","AAPL","MSFT","GOOGL","GOOG","NVDA","MU","AMD","INTC","MRVL","QCOMM","SQ","PYPL","TTD","TSLA","ZM","TDOC","LVGO","MDB","HD","VNQ","ARI","ACC","IIPR","EQR","EPR","SPG","PLD","ACB","WHR","NVAX","APT","MDT","CLRX","COST","SDC","LK","PVH","KSS","M","LULU","NKE","KO","BAC","JPM","CS","WFC","ARKW","ARKK","MGM","AMAT","WYNN","TGT","ITT","FXI"]
with ThreadPoolExecutor() as executor:
for ticker in symbols:
downloadChain(ticker)
第二个更标准。我们在其中定义了一个 async main，我们要求 asyncio 将其用作主入口点。
import asyncio
import pandas as pd
import yfinance as yf
from concurrent.futures import ProcessPoolExecutor
symbols = ["WATT","TSLA","UVXY","VXX","KEYS","EGO","GLD","WORK","BYND","BLK","PINS","LYFT","SPCE","PAYC","WDAY","UBER","CHGG","SHAK","CMG","CTL","ACB","TLRY","CGC","MJ","ORCL","GRUB","RNG","JWN","TTWO","ADI","ATVI","EA","SNE","GAMR","TXN","TMUS","MCHP","TSM","XBI","ETFC","MS","IWM","EXPD","RCL","CCL","MOMO","BABA","VMW","CRM","ULTA","SKYY","SPLK","FLWS","AVGO","TWTR","PANW","RJF","SABR","LOW","RS","ON","VEEV","DOCU","FB","SNAP","HPQ","RACE","F","AMAT","MRO","STM","AAL","DAL","VICR","XLC","CRON","DELL","T","VZ","S","MELI","CVM","REGN","NVAX","APT","CODX","LAKE","MRNA","EBS","INO", "SPY","SH","QQQ","XLF","KRE","XLV","HYG","LQD","NET","NFLX","ROKU","SHOP","AMZN","AAPL","MSFT","GOOGL","GOOG","NVDA","MU","AMD","INTC","MRVL","QCOMM","SQ","PYPL","TTD","TSLA","ZM","TDOC","LVGO","MDB","HD","VNQ","ARI","ACC","IIPR","EQR","EPR","SPG","PLD","ACB","WHR","NVAX","APT","MDT","CLRX","COST","SDC","LK","PVH","KSS","M","LULU","NKE","KO","BAC","JPM","CS","WFC","ARKW","ARKK","MGM","AMAT","WYNN","TGT","ITT","FXI"]
done = []
def downloadChain(ticker):
print(ticker)
df = pd.DataFrame()
daysOut = 100
chain = 0
try:
yf_ticker = yf.Ticker(ticker)
expiration_dates = yf_ticker.options
for expiration_date in expiration_dates:
if (datetime.fromisoformat(expiration_date) - datetime.now()).days <= daysOut:
try:
chain = yf_ticker.option_chain(expiration_date)
df = df.append(chain)
except Exception as e:
pass
except Exception as e:
pass
done.append(ticker)
async def main():
with ProcessPoolExecutor() as executor:
for ticker in symbols:
asyncio.get_event_loop().run_in_executor(executor, downloadChain,
ticker)
if __name__ == '__main__':
asyncio.run(main())
在这里，您还可以更精细地控制要使用的执行程序。基本上，我们明确地编码我们正在使用的事件循环以及我们将工作添加到执行程序的事件循环。 ProcessPoolExecutor 和 ThreadPoolExecutor.
之间的本地测试没有显示出很大的差异
我目前在使用 Joblib 运行 多处理或并行程序时遇到问题。我之前能够让它工作，并且我总共达到了 1 分钟的时间，但是，我四处走动并改变了很多，并且弄乱了一些东西。我已经发布了准系统代码，因为我收到了同样的错误。我正在尝试遍历所有 150 个股票代码，并使用 yahoo finance 来接收每个股票代码的期权链。我正在尝试在一分钟内完成此操作。我也尝试过其他库，比如 asyncio，但没有成功。任何建议将不胜感激。
import yfinance as yf
def background(f):
def wrapped(*args, **kwargs):
return asyncio.get_event_loop().run_in_executor(None, f, *args, **kwargs)
return wrapped
done = []
@background
def downloadChain(ticker):
print(ticker)
df = pd.DataFrame()
daysOut = 100
chain = 0
try:
yf_ticker = yf.Ticker(ticker)
expiration_dates = yf_ticker.options
for expiration_date in expiration_dates:
if (datetime.fromisoformat(expiration_date) - datetime.now()).days <= daysOut:
try:
chain = yf_ticker.option_chain(expiration_date)
df = df.append(chain)
except Exception as e:
pass
except Exception as e:
pass
done.append(ticker)
主要功能：
symbols = ["WATT","TSLA","UVXY","VXX","KEYS","EGO","GLD","WORK","BYND","BLK","PINS","LYFT","SPCE","PAYC","WDAY","UBER","CHGG","SHAK","CMG","CTL","ACB","TLRY","CGC","MJ","ORCL","GRUB","RNG","JWN","TTWO","ADI","ATVI","EA","SNE","GAMR","TXN","TMUS","MCHP","TSM","XBI","ETFC","MS","IWM","EXPD","RCL","CCL","MOMO","BABA","VMW","CRM","ULTA","SKYY","SPLK","FLWS","AVGO","TWTR","PANW","RJF","SABR","LOW","RS","ON","VEEV","DOCU","FB","SNAP","HPQ","RACE","F","AMAT","MRO","STM","AAL","DAL","VICR","XLC","CRON","DELL","T","VZ","S","MELI","CVM","REGN","NVAX","APT","CODX","LAKE","MRNA","EBS","INO", "SPY","SH","QQQ","XLF","KRE","XLV","HYG","LQD","NET","NFLX","ROKU","SHOP","AMZN","AAPL","MSFT","GOOGL","GOOG","NVDA","MU","AMD","INTC","MRVL","QCOMM","SQ","PYPL","TTD","TSLA","ZM","TDOC","LVGO","MDB","HD","VNQ","ARI","ACC","IIPR","EQR","EPR","SPG","PLD","ACB","WHR","NVAX","APT","MDT","CLRX","COST","SDC","LK","PVH","KSS","M","LULU","NKE","KO","BAC","JPM","CS","WFC","ARKW","ARKK","MGM","AMAT","WYNN","TGT","ITT","FXI"]
for ticker in symbols:
downloadChain(ticker)
我添加了一个单独的循环来查看 "done" 数组的大小，它包含所有已完成的符号。我不确定我做了什么更改，但现在这个循环将在大约 10-15 分钟内完成，而预计需要 1 分钟。
while True:
clear_output(wait=True)
print(len(done))
"fix"有两个版本。将它们添加为答案而不是将评论用作聊天 :)
import asyncio
import pandas as pd
import yfinance as yf
from concurrent.futures import ThreadPoolExecutor
def background(f):
def wrapped(*args, **kwargs):
return asyncio.get_event_loop().run_in_executor(executor, f, *args, **kwargs)
return wrapped
done = []
@background
def downloadChain(ticker):
print(ticker)
df = pd.DataFrame()
daysOut = 100
chain = 0
try:
yf_ticker = yf.Ticker(ticker)
expiration_dates = yf_ticker.options
for expiration_date in expiration_dates:
if (datetime.fromisoformat(expiration_date) - datetime.now()).days <= daysOut:
try:
chain = yf_ticker.option_chain(expiration_date)
df = df.append(chain)
except Exception as e:
pass
except Exception as e:
pass
done.append(ticker)
symbols = ["WATT","TSLA","UVXY","VXX","KEYS","EGO","GLD","WORK","BYND","BLK","PINS","LYFT","SPCE","PAYC","WDAY","UBER","CHGG","SHAK","CMG","CTL","ACB","TLRY","CGC","MJ","ORCL","GRUB","RNG","JWN","TTWO","ADI","ATVI","EA","SNE","GAMR","TXN","TMUS","MCHP","TSM","XBI","ETFC","MS","IWM","EXPD","RCL","CCL","MOMO","BABA","VMW","CRM","ULTA","SKYY","SPLK","FLWS","AVGO","TWTR","PANW","RJF","SABR","LOW","RS","ON","VEEV","DOCU","FB","SNAP","HPQ","RACE","F","AMAT","MRO","STM","AAL","DAL","VICR","XLC","CRON","DELL","T","VZ","S","MELI","CVM","REGN","NVAX","APT","CODX","LAKE","MRNA","EBS","INO", "SPY","SH","QQQ","XLF","KRE","XLV","HYG","LQD","NET","NFLX","ROKU","SHOP","AMZN","AAPL","MSFT","GOOGL","GOOG","NVDA","MU","AMD","INTC","MRVL","QCOMM","SQ","PYPL","TTD","TSLA","ZM","TDOC","LVGO","MDB","HD","VNQ","ARI","ACC","IIPR","EQR","EPR","SPG","PLD","ACB","WHR","NVAX","APT","MDT","CLRX","COST","SDC","LK","PVH","KSS","M","LULU","NKE","KO","BAC","JPM","CS","WFC","ARKW","ARKK","MGM","AMAT","WYNN","TGT","ITT","FXI"]
with ThreadPoolExecutor() as executor:
for ticker in symbols:
downloadChain(ticker)
第二个更标准。我们在其中定义了一个 async main，我们要求 asyncio 将其用作主入口点。
import asyncio
import pandas as pd
import yfinance as yf
from concurrent.futures import ProcessPoolExecutor
symbols = ["WATT","TSLA","UVXY","VXX","KEYS","EGO","GLD","WORK","BYND","BLK","PINS","LYFT","SPCE","PAYC","WDAY","UBER","CHGG","SHAK","CMG","CTL","ACB","TLRY","CGC","MJ","ORCL","GRUB","RNG","JWN","TTWO","ADI","ATVI","EA","SNE","GAMR","TXN","TMUS","MCHP","TSM","XBI","ETFC","MS","IWM","EXPD","RCL","CCL","MOMO","BABA","VMW","CRM","ULTA","SKYY","SPLK","FLWS","AVGO","TWTR","PANW","RJF","SABR","LOW","RS","ON","VEEV","DOCU","FB","SNAP","HPQ","RACE","F","AMAT","MRO","STM","AAL","DAL","VICR","XLC","CRON","DELL","T","VZ","S","MELI","CVM","REGN","NVAX","APT","CODX","LAKE","MRNA","EBS","INO", "SPY","SH","QQQ","XLF","KRE","XLV","HYG","LQD","NET","NFLX","ROKU","SHOP","AMZN","AAPL","MSFT","GOOGL","GOOG","NVDA","MU","AMD","INTC","MRVL","QCOMM","SQ","PYPL","TTD","TSLA","ZM","TDOC","LVGO","MDB","HD","VNQ","ARI","ACC","IIPR","EQR","EPR","SPG","PLD","ACB","WHR","NVAX","APT","MDT","CLRX","COST","SDC","LK","PVH","KSS","M","LULU","NKE","KO","BAC","JPM","CS","WFC","ARKW","ARKK","MGM","AMAT","WYNN","TGT","ITT","FXI"]
done = []
def downloadChain(ticker):
print(ticker)
df = pd.DataFrame()
daysOut = 100
chain = 0
try:
yf_ticker = yf.Ticker(ticker)
expiration_dates = yf_ticker.options
for expiration_date in expiration_dates:
if (datetime.fromisoformat(expiration_date) - datetime.now()).days <= daysOut:
try:
chain = yf_ticker.option_chain(expiration_date)
df = df.append(chain)
except Exception as e:
pass
except Exception as e:
pass
done.append(ticker)
async def main():
with ProcessPoolExecutor() as executor:
for ticker in symbols:
asyncio.get_event_loop().run_in_executor(executor, downloadChain,
ticker)
if __name__ == '__main__':
asyncio.run(main())
在这里，您还可以更精细地控制要使用的执行程序。基本上，我们明确地编码我们正在使用的事件循环以及我们将工作添加到执行程序的事件循环。 ProcessPoolExecutor 和 ThreadPoolExecutor.