twstock stock.py change rand sleep time to 3-10 sec for request
stock.py change rand sleep time to 3-10 sec for request to avoid bantock.py source as below:
在 twstock/stock.py#L57 的下一行
for 迴圈裏面 增加 以下兩行 code 即可import time
import random
# -*- coding: utf-8 -*-
import datetime
import time
import urllib.parse
from collections import namedtuple
import random
from twstock.proxy import get_proxies
try:
from json.decoder import JSONDecodeError
except ImportError:
JSONDecodeError = ValueError
import requests
try:
from . import analytics
from .codes import codes
except ImportError as e:
if e.name == 'lxml':
# Fix #69
raise e
import analytics
from codes import codes
TWSE_BASE_URL = 'http://www.twse.com.tw/'
TPEX_BASE_URL = 'http://www.tpex.org.tw/'
DATATUPLE = namedtuple('Data', ['date', 'capacity', 'turnover', 'open',
'high', 'low', 'close', 'change', 'transaction'])
class BaseFetcher(object):
def fetch(self, year, month, sid, retry):
pass
def _convert_date(self, date):
"""Convert '106/05/01' to '2017/05/01'"""
return '/'.join() + 1911)] + date.split('/'))
def _make_datatuple(self, data):
pass
def purify(self, original_data):
pass
class TWSEFetcher(BaseFetcher):
REPORT_URL = urllib.parse.urljoin(
TWSE_BASE_URL, 'exchangeReport/STOCK_DAY')
def __init__(self):
pass
def fetch(self, year: int, month: int, sid: str, retry: int=5):
params = {'date': '%d%02d01' % (year, month), 'stockNo': sid}
for retry_i in range(retry):
randNum = random.randrange(3,10)
time.sleep(randNum)
r = requests.get(self.REPORT_URL, params=params,
proxies=get_proxies())
try:
data = r.json()
except JSONDecodeError:
continue
else:
break
else:
# Fail in all retries
data = {'stat': '', 'data': []}
if data['stat'] == 'OK':
data['data'] = self.purify(data)
else:
data['data'] = []
return data
def _make_datatuple(self, data):
data = datetime.datetime.strptime(
self._convert_date(data), '%Y/%m/%d')
data = int(data.replace(',', ''))
data = int(data.replace(',', ''))
data = None if data == '--' else float(data.replace(',', ''))
data = None if data == '--' else float(data.replace(',', ''))
data = None if data == '--' else float(data.replace(',', ''))
data = None if data == '--' else float(data.replace(',', ''))
# +/-/X表示漲/跌/不比價
data = float(0.0 if data.replace(',', '') ==
'X0.00' else data.replace(',', ''))
data = int(data.replace(',', ''))
return DATATUPLE(*data)
def purify(self, original_data):
return ]
class TPEXFetcher(BaseFetcher):
REPORT_URL = urllib.parse.urljoin(TPEX_BASE_URL,
'web/stock/aftertrading/daily_trading_info/st43_result.php')
def __init__(self):
pass
def fetch(self, year: int, month: int, sid: str, retry: int=5):
params = {'d': '%d/%d' % (year - 1911, month), 'stkno': sid}
for retry_i in range(retry):
r = requests.get(self.REPORT_URL, params=params,
proxies=get_proxies())
try:
data = r.json()
except JSONDecodeError:
continue
else:
break
else:
# Fail in all retries
data = {'aaData': []}
data['data'] = []
if data['aaData']:
data['data'] = self.purify(data)
return data
def _convert_date(self, date):
"""Convert '106/05/01' to '2017/05/01'"""
return '/'.join() + 1911)] + date.split('/'))
def _make_datatuple(self, data):
data = datetime.datetime.strptime(self._convert_date(data.replace('＊', '')),
'%Y/%m/%d')
data = int(data.replace(',', '')) * 1000
data = int(data.replace(',', '')) * 1000
data = None if data == '--' else float(data.replace(',', ''))
data = None if data == '--' else float(data.replace(',', ''))
data = None if data == '--' else float(data.replace(',', ''))
data = None if data == '--' else float(data.replace(',', ''))
data = float(data.replace(',', ''))
data = int(data.replace(',', ''))
return DATATUPLE(*data)
def purify(self, original_data):
return ]
class Stock(analytics.Analytics):
def __init__(self, sid: str, initial_fetch: bool=True):
self.sid = sid
self.fetcher = TWSEFetcher(
) if codes.market == '上市' else TPEXFetcher()
self.raw_data = []
self.data = []
# Init data
if initial_fetch:
self.fetch_31()
def _month_year_iter(self, start_month, start_year, end_month, end_year):
ym_start = 12 * start_year + start_month - 1
ym_end = 12 * end_year + end_month
for ym in range(ym_start, ym_end):
y, m = divmod(ym, 12)
yield y, m + 1
def fetch(self, year: int, month: int):
"""Fetch year month data"""
self.raw_data =
self.data = self.raw_data['data']
return self.data
def fetch_from(self, year: int, month: int):
"""Fetch data from year, month to current year month data"""
self.raw_data = []
self.data = []
today = datetime.datetime.today()
for year, month in self._month_year_iter(month, year, today.month, today.year):
start = time.time()
self.raw_data.append(self.fetcher.fetch(year, month, self.sid))
self.data.extend(self.raw_data[-1]['data'])
done = time.time()
elapsed = int(done - start)
print('fetch from {}/{}, time:{}'.format(year, month, elapsed), end=' ')
return self.data
def fetch_31(self):
"""Fetch 31 days data"""
today = datetime.datetime.today()
before = today - datetime.timedelta(days=60)
self.fetch_from(before.year, before.month)
self.data = self.data[-31:]
return self.data
@property
def date(self):
return
@property
def capacity(self):
return
@property
def turnover(self):
return
@property
def price(self):
return
@property
def high(self):
return
@property
def low(self):
return
@property
def open(self):
return
@property
def close(self):
return
@property
def change(self):
return
@property
def transaction(self):
return
https://github.com/joytsay/twstock/blob/master/twstock/stock.py">twstock/stock.py at master · joytsay/twstock · GitHub
頁:
[1]