My2PhaseCrawler.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "### Two in one\n",
    "##\n",
    "\n",
    "#!/usr/bin/python\n",
    "# -*- coding: utf-8 -*-\n",
    "\n",
    "import os\n",
    "import json\n",
    "import csv\n",
    "import time\n",
    "\n",
    "from datetime import date\n",
    "\n",
    "import requests\n",
    "\n",
    "class CrawlerController(object):\n",
    "    '''Split targets into several Crawler, avoid request url too long'''\n",
    "\n",
    "    def __init__(self, targets, max_stock_per_crawler=50):\n",
    "        self.crawlers = []\n",
    "\n",
    "        for index in range(0, len(targets), max_stock_per_crawler):\n",
    "            crawler = Crawler(targets[index:index + max_stock_per_crawler])\n",
    "            self.crawlers.append(crawler)\n",
    "\n",
    "    def run(self):\n",
    "        data = []\n",
    "        for crawler in self.crawlers:\n",
    "            data.extend(crawler.get_data())\n",
    "        return data\n",
    "\n",
    "class Crawler(object):\n",
    "    '''Request to Market Information System'''\n",
    "    def __init__(self, targets):\n",
    "        endpoint = 'http://mis.twse.com.tw/stock/api/getStockInfo.jsp'\n",
    "        # Add 1000 seconds for prevent time inaccuracy\n",
    "        timestamp = int(time.time() * 1000 + 1000000)\n",
    "        ## channels = '|'.join('tse_{}.tw'.format(target) for target in targets)\n",
    "        ##  please mark tse_|otc_ markets in stocknumber.csv \n",
    "        channels = '|'.join('{}.tw'.format(target) for target in targets)\n",
    "        self.query_url = '{}?_={}&ex_ch={}'.format(endpoint, timestamp, channels)\n",
    "\n",
    "    def get_data(self):\n",
    "        try:\n",
    "            # Get original page to get session\n",
    "            req = requests.session()\n",
    "            req.get('http://mis.twse.com.tw/stock/index.jsp',\n",
    "                    headers={'Accept-Language': 'zh-TW'})\n",
    "\n",
    "            response = req.get(self.query_url)\n",
    "            content = json.loads(response.text)\n",
    "        except Exception as err:\n",
    "            print(err)\n",
    "            data = []\n",
    "        else:\n",
    "            data = content['msgArray']\n",
    "\n",
    "        return data\n",
    "\n",
    "class Recorder(object):\n",
    "    '''Record data to csv'''\n",
    "    def __init__(self, path='data'):\n",
    "        self.folder_path = '{}/{}'.format(path, date.today().strftime('%Y%m%d'))\n",
    "        if not os.path.isdir(self.folder_path):\n",
    "            os.mkdir(self.folder_path)\n",
    "\n",
    "    def record_to_csv(self, data):\n",
    "        for row in data:\n",
    "            try:\n",
    "                file_path = '{}/{}.csv'.format(self.folder_path, row['c'])\n",
    "                print (\"File :\",file_path)\n",
    "                with open(file_path, 'a') as output_file:\n",
    "                    writer = csv.writer(output_file, delimiter=',', quotechar='\"')\n",
    "                    writer.writerow([\n",
    "                        row['t'], # 資料時間\n",
    "                        row['z'], # 最近成交價\n",
    "                        row['tv'],# 當盤成交量\n",
    "                        row['v'], # 當日累計成交量\n",
    "                        row['a'], # 最佳五檔賣出價格\n",
    "                        row['f'], # 最價五檔賣出數量\n",
    "                        row['b'], # 最佳五檔買入價格\n",
    "                        row['g'], # 最佳五檔買入數量\n",
    "                        row['n'], # name \n",
    "                        row['o'], # open \n",
    "                        row['l'], # low \n",
    "                        row['h'], # high \n",
    "                        row['y']  # \n",
    "                    ])\n",
    "                    output_file.flush() # whenever you want, and/or\n",
    "                #output_file.close() # when you're done.\n",
    "            except Exception as err:\n",
    "                print('Fatal: ',err)\n",
    "\n",
    "import datetime\n",
    "import requests\n",
    "import sched\n",
    "import time as tm\n",
    "import json\n",
    "from time import gmtime, strftime                \n",
    "                \n",
    "s = sched.scheduler(tm.time, tm.sleep)\n",
    "\n",
    "from Stock import stock_mrk \n",
    "\n",
    "def main_crawler ():  \n",
    "    tar   =  [(_.strip()) for _ in open('stocknumber.csv', 'r')] # Append TSE OTC market to stock id\n",
    "    \n",
    "    targets = [stock_mrk(_.strip()) for _ in open('stocknumber.csv', 'r')] # Append TSE OTC market to stock id\n",
    "    \n",
    "    print (tar, \" + \", targets)\n",
    "    \n",
    "    time = datetime.datetime.now()  \n",
    "    print(\"開始更新時間:\" + str(time.date())+':'+str(time.hour)+\":\"+str(time.minute)+\":\"+str(time.second))\n",
    "\n",
    "    start_time = datetime.datetime.strptime(str(time.date())+'9:00', '%Y-%m-%d%H:%M')\n",
    "    end_time =  datetime.datetime.strptime(str(time.date())+'22:31', '%Y-%m-%d%H:%M')\n",
    "    \n",
    "    # tm.sleep (3) # 避免證交所伺服器鎖 IP，可能為都是網頁伺服器的rate limiting 在作祟。\n",
    "    # 判斷爬蟲終止條件\n",
    "    sleeptimer = 0.5\n",
    "    if time >= start_time and time <= end_time:\n",
    "        tm.sleep (sleeptimer)\n",
    "        try: \n",
    "            controller = CrawlerController(targets)\n",
    "            data = controller.run()\n",
    "\n",
    "            recorder = Recorder()\n",
    "            recorder.record_to_csv(data)\n",
    "        except:\n",
    "            msg = \"證交所網路忙碌！\"\n",
    "        else:\n",
    "            msg = \"資料擷取時間:\" + str(time.date())+':'+str(time.hour)+\":\"+str(time.minute)+\":\"+str(time.second)\n",
    "        finally:\n",
    "            # print(\"更新時間:\" + str(time.date())+':'+str(time.hour)+\":\"+str(time.minute)+\":\"+str(time.second))\n",
    "            print(\"Done.\", msg)\n",
    "            s.enter(1, 0, main_crawler, argument=())\n",
    "    else:\n",
    "        print ('非營業時間，不提供連續資料。')\n",
    "        print ('繼續等待交易時間。。。')\n",
    "        s.enter(1, 0, main_crawler, argument=())        \n",
    "                \n",
    "\n",
    "### \n",
    "#   Regenerate purged price file\n",
    "### \n",
    "\n",
    "\n",
    "# 讀取 CSV File\n",
    "\n",
    "def RegenController(targets):\n",
    "    \n",
    "    import pandas as pd # 引用套件並縮寫為 pd  \n",
    "\n",
    "    import csv\n",
    "    import os\n",
    "    from os import listdir\n",
    "    from os.path import isfile, join\n",
    "    from datetime import date\n",
    "\n",
    "    print ('股票代碼: ',targets)\n",
    "    today = str(date.today().year).zfill(4)+str(date.today().month).zfill(2)+str(date.today().day).zfill(2)\n",
    "    ## Template today string for test\n",
    "    # today = '20200903'\n",
    "\n",
    "    # 今天抓到的清單\n",
    "    index_list = [ f[:-4] for f in listdir(join('data', today)) if f[-4:] == '.csv' ]\n",
    "\n",
    "    # 刪除重複的資料並重新排序\n",
    "    for stock_id in index_list:\n",
    "        f = open(join('data', today, stock_id+'.csv'), 'rt')\n",
    "\n",
    "        colnames=['時間', '即時價位', '即時量', '總量', '5檔賣價', '5檔賣量', '5檔買價', '5檔買量', 'name', 'open', 'low', 'high', 'last'] \n",
    "        df = pd.read_csv(f,names=colnames, header=None)  # header=None, usecols=[0,1,2,3,4,5,6,7])\n",
    "      \n",
    "        result_df = df.drop_duplicates(subset=['時間', '總量'], keep='first')\n",
    "        result_df.reset_index(inplace=False)\n",
    " \n",
    "        result_df['Sale5']  = df['5檔賣價'].str.split('_')  \n",
    "        result_df['Sale5V'] = df['5檔賣量'].str.split('_')\n",
    "        result_df['Buy5']   = df['5檔買價'].str.split('_')\n",
    "        result_df['Buy5V']  = df['5檔買量'].str.split('_')\n",
    "\n",
    "        result_df = result_df.drop(columns=['5檔賣價', '5檔賣量', '5檔買價', '5檔買量'])\n",
    "        result_df.reset_index(inplace=True)\n",
    "\n",
    "        last_v = 0 \n",
    "        last_p = result_df.loc[0,'Sale5'][0]\n",
    "\n",
    "        for i in range(0, len(result_df)): \n",
    "            if int(result_df.loc[i,'總量']) <= last_v  :\n",
    "                result_df = result_df.drop([i])\n",
    "\n",
    "                continue \n",
    "            else: \n",
    "                if (str(result_df.loc[i,'即時價位']) == '-'):\n",
    "                    result_df.loc[i, '即時價位'] = last_p\n",
    "                else:\n",
    "                    last_p = float(result_df.loc[i,'即時價位'])\n",
    "                if (str(result_df.loc[i,'即時量']) == '-'):\n",
    "                    result_df.loc[i, '即時量'  ] = int(result_df.loc[i, '總量']) -  last_v\n",
    "                last_v = result_df.loc[i,'總量']  \n",
    "\n",
    "        result_df.reset_index(inplace=True)\n",
    "\n",
    "\n",
    "        # print ('Final ',result_df.head(10))\n",
    "        f = join('data', today, stock_id+'._csv')  # template CSV File \n",
    "        result_df.to_csv(f, header=0,index=False) \n",
    "\n",
    "    print ('All Done! ') \n",
    "# \n",
    "\n",
    "\n",
    "#\n",
    "#  Main loop \n",
    "#\n",
    "import datetime\n",
    "import requests\n",
    "import sched\n",
    "import time as tm\n",
    "import json\n",
    "from time import gmtime, strftime                \n",
    "\n",
    "\n",
    "def Regen_loop ():        \n",
    "    targets = [_.strip() for _ in open('stocknumber.csv', 'r')]\n",
    "    \n",
    "    time = datetime.datetime.now()  \n",
    "    print(\"開始統整時間:\" + str(time.date())+':'+str(time.hour)+\":\"+str(time.minute)+\":\"+str(time.second))\n",
    "\n",
    "    start_time = datetime.datetime.strptime(str(time.date())+'9:00', '%Y-%m-%d%H:%M')\n",
    "    end_time =  datetime.datetime.strptime(str(time.date())+'22:31', '%Y-%m-%d%H:%M')\n",
    "    \n",
    "    # tm.sleep (3) # 避免證交所伺服器鎖 IP，可能為都是網頁伺服器的rate limiting 在作祟。\n",
    "    # 判斷爬蟲終止條件\n",
    "    sleeptimer = 5.5\n",
    "    if time >= start_time and time <= end_time:\n",
    "        tm.sleep (sleeptimer)\n",
    "        try: \n",
    "            RegenController(targets)\n",
    "            \n",
    "        except:\n",
    "            msg = \"證交所網路忙碌！\"\n",
    "        else:\n",
    "            msg = \"_csv 更新時間:\" + str(time.date())+':'+str(time.hour)+\":\"+str(time.minute)+\":\"+str(time.second)\n",
    "        finally:\n",
    "            # print(\"更新時間:\" + str(time.date())+':'+str(time.hour)+\":\"+str(time.minute)+\":\"+str(time.second))\n",
    "            print(\"Done...\", msg)\n",
    "            s.enter(5, 0, Regen_loop, argument=())\n",
    "    else:\n",
    "        print ('非營業時間，不須統整。')\n",
    "        print ('繼續等待交易時間。。。')\n",
    "        s.enter(5, 0, Regen_loop, argument=())        \n",
    "            \n",
    "            \n",
    "s = sched.scheduler(tm.time, tm.sleep)            \n",
    "if __name__ == '__main__':\n",
    "    \n",
    "    s.enter(1, 0, main_crawler, argument=())\n",
    "    s.enter(5, 0, Regen_loop, argument=())\n",
    "    s.run()\n"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.9.0"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}