Python爬蟲爬取上海黃金交易所歷史交易數據

為什么

因為想做上海黃金的量化交易，又信不過網上那些忽悠的神乎其神的App。於是自己動手，豐衣足食。

如何做

首先要獲取權威的交易數據，上海黃金交易所官網就有歷年的交易數據。所以打算用熟悉的Python寫個爬蟲自動獲取。

1. 工具准備

Python3.6 + requests + lxml + Json

2. 網站解析

首先找到上海黃金交易所每日行情頁列表（首頁 > 數據資訊 > 歷史行情數據 > 每日行情），分析該列表每頁顯示10天
的數據列表，點開后才是每天每個交易合約的交易數據。並且每一頁的URL采用參數方式進行定位，如：“sjzx/mrhqsj?p=2 ”
表示第二頁。所以只需要一個簡單循環就可以找到需要的頁面。

其次要找到具體數據頁面列表的Xpath，可以使用瀏覽器Chrome自帶的開發者模式，找到需要的數據，直接點右鍵 Copy > Copy Xpath。

3. 上代碼

# -*- coding: UTF-8 -*-
# 本模塊從上海黃金交易所官網下載歷史交易數據
# https://www.sge.com.cn/sjzx/mrhqsj

import os
import time
from Lib.Web import get_Html, get_list, get_List_xpath, add_host
from Lib.os import save_list, save_list_A, save_list_B, makdir, BASE_PATH

headers = {
    'Referer': 'https://www.sge.com.cn/sjzx/mrhqsj',
    'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/87.0.4280.88 Safari/537.36',
}


def get_table(title, url, headers):
    table_xpath = '//div[@class="content"]/table/tbody/tr/td[1]/text()'
    html = get_Html(url, headers)
    doc = get_list(html, table_xpath)
    tab = []
    had = []
    n = len(doc)
    for r in range(1, n+1):
        table_xpath = '//div[@class="content"]/table/tbody/tr[%d]/td/text()' % r
        d = get_list(html, table_xpath)
        if r == 1:
            for i in d:
                had.append(str(i).replace('\t', '').replace(
                    '\n', '').replace('\r', ''))
        else:
            row = {}
            row['交易日期'] = title
            try:
                for i in range(len(d)):
                    row[had[i]] = str(d[i]).replace(
                        '\t', '').replace('\n', '').replace('\r', '')
            except Exception as e:
                pass
            tab.append(row)
    return tab


if __name__ == "__main__":

    # 獲得下載鏈接
    for r in range(1, 201):
        url = 'https://www.sge.com.cn/sjzx/mrhqsj?p=%d' % r
        filename = 'list_%d.txt' % r
        cache_dir = "goldlist"

        html = get_Html(url, headers)

        if not os.path.exists(os.path.join(BASE_PATH, cache_dir)):
            makdir(os.path.join(BASE_PATH, cache_dir))

        filename = os.path.join(BASE_PATH, cache_dir, filename)
        if os.path.exists(filename):
            print("跳過：%s" % filename)
            continue

        a = '/html/body/div[6]/div/div[2]/div[2]/div[2]/ul/li/a/span[2]/text()'
        b = '/html/body/div[6]/div/div[2]/div[2]/div[2]/ul/li/a/@href'
        lst = get_List_xpath(html, a, b)
        for item in lst:
            lst[item] = add_host(url, lst[item])

        save_list_A(filename, lst)

        print('獲取歷史行情第%d頁' % r)

        time.sleep(3)

    # 下載行情數據
    for r in range(1, 201):
        url = 'https://www.sge.com.cn/sjzx/mrhqsj?p=%d' % r
        filename = 'list_%d.txt' % r
        cache_dir = "goldlist"
        filename = os.path.join(BASE_PATH, cache_dir, filename)
        if os.path.exists(filename):
            with open(filename, 'r', encoding='utf-8') as f:
                line = f.readline()
                item, url = line.split('\t')
                filename = os.path.join(BASE_PATH, cache_dir, "%s.txt" % item)

                if os.path.exists(filename):
                    print("跳過：%s" % filename)
                    continue

                doc = get_table(item, str(url).replace('\n', ''), headers)
                save_list_B(filename, doc)

                print("保存：%s" % filename)

                time.sleep(3)

其中使用到我自己為了方便建立的庫函數

def get_host(url):
    """ 返回域名，如：https://www.baidu.com """
    ul = urlparse(url)
    return ul.scheme + '://' + ul.hostname


def add_host(url, path):
    return get_host(url) + path


def get_Html(url, headers, cookies=None, params=None):
    """ 返回網頁內容 """
    if cookies:
        r = requests.get(url=url, headers=headers, cookies=cookies)
    else:
        r = requests.get(url=url, headers=headers)
    r.encoding = "utf-8"
    return etree.HTML(r.text)


def get_list(html, xpath):
    """ 返回指定位置的列表 """
    return html.xpath(xpath)


def save_list_B(filename, list):
    with open(filename, 'w', encoding='utf-8') as f:
        f.writelines(json.dumps(list,ensure_ascii=False))