-
Notifications
You must be signed in to change notification settings - Fork 6
/
Copy pathutils.py
74 lines (65 loc) · 2.37 KB
/
utils.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
import os
import re
import time
import requests
from loguru import logger
from datetime import timedelta
from datetime import datetime as dt
def GetOptionCode():
logger.info('Retrieving option code list.')
# Find all date
url = 'https://tw.screener.finance.yahoo.net/future/aa03?opmr=optionfull&opcm=WTXO'
with requests.get(url, stream=True) as r:
tag = re.escape('</option>')
p = re.compile(r'([\dW]+)%s' % tag)
date = p.findall(r.text)
# Find all price in each date
code_list = []
for d in date:
y = int(d[:4]) % 10
m = int(d[4:6]) - 1
w = d[6:]
if not w: w = 'OO'
buy = 'ABCDEFGHIJKL'
sell = 'MNOPQRSTUVWX'
url = f'https://tw.screener.finance.yahoo.net/future/aa03?opmr=optionfull&opcm=WTXO&opym={d}'
tag = re.escape('<td class="ext-big-tb-center">')
p = re.compile(r'%s(\d+)' % tag)
with requests.get(url) as r:
prices = p.findall(r.text)
a = int(prices[0])
b = int(prices[-1])
logger.info(f'{d:<8s} price range ({a:5}, {b:5}), middle: {(a+b)//2}.')
for p in prices:
code_list.append(f'TX{w[1]}{int(p):05d}{buy[m]}{y}')
code_list.append(f'TX{w[1]}{int(p):05d}{sell[m]}{y}')
return code_list
def walk(path):
for dir_path, _, file_list in os.walk(path):
for file_name in file_list:
full_path = os.path.join(dir_path, file_name)
yield full_path
def remove_dir(dir_path):
for full_path in walk(dir_path):
os.remove(full_path)
os.removedirs(dir_path)
def clear_log(path):
for dir_path, dir_list, file_list in os.walk(path):
date = dt.now().strftime('%Y%m%d')
date2 = (dt.now() - timedelta(days=1)).strftime('%Y%m%d')
for dir_name in dir_list:
if dir_name != date and dir_name != date2:
remove_dir(os.path.join(dir_path, dir_name))
for file_name in file_list:
full_path = os.path.join(dir_path, file_name)
flag = True
while flag:
try:
open(full_path, 'w').close()
flag = False
except Exception as e:
logger.debug(f'{full_path} remove failed, will retry after 1 second.')
time.sleep(1)
if __name__ == '__main__':
GetOptionCode()
clear_log('./Logs')