tushare宏觀數據使用pandas入庫，增加唯一索引

1，對pandas入數據庫

pandas代碼中自帶了to_sql的方法可以直接使用。
但是數據字段是text的，需要修改成 varchar的，否則不能增加索引。
在增加了主鍵之後可以控製數據不能再增量修改了。

增加：dtype類型可以解決

dtype={col_name: NVARCHAR(length=255) 
    for col_name in data.columns.tolist()}

to_sql 中帶的 if_exists 是針對不存在的數據表操作的。而不是數據。
查看了pandas io 下麵的 sql.py

2，分析下入庫

sql.py 使用了SQLAlchemy 進行數據存儲。分為普通SQLDatabase和SQLiteTable兩個實現類，實現操作 read_sql 和 to_sql 兩個方法。

實現了數據庫的讀取和數據庫的入庫。

思路本來是想找到入庫的SQL，發現隻是在SQLiteTable 才拚接了INSERT INTO 。在操作mysql的時候根本沒有調用。而是直接使用 SQLAlchemy 進行麵向對象入庫的。

目標是拚接下 INSERT IGNORE INTO ，支持增量插入數據。
並增加 PRIMARY KEY，數據不被重複插入。

最後找到一個方法，直接修改sqlalchemy 下 mysql的代碼：

sed -i -e 's/executemany(statement/executemany(statement.replace\("INSERT INTO","INSERT IGNORE INTO")/g' \
        /usr/local/lib/python3.5/dist-packages/sqlalchemy/dialects/mysql/mysqldb.py 

即可解決重複入庫的問題。
也就是把 “INSERT INTO” 替換成 “INSERT IGNORE INTO” 的方式。
讓數據庫不保存，保證數據更新成功。

非常簡單暴力的方式。

3，增加唯一索引

增加索引很簡單：

ALTER IGNORE TABLE `ts_deposit_rate` ADD PRIMARY KEY (`date`,`deposit_type`);

直接可以增加，但是數據在增加完索引之後，就不能重複再增加了。
數據庫會報錯誤。就需要對數據庫的表的索引進行判斷了。

還好 sqlalchemy 的 inspect 功能非常的強大。
可以直接獲得表的 primary key。再判斷下就好了。

#定義engine
engine = common.engine()
#使用 https://docs.sqlalchemy.org/en/latest/core/reflection.html
#使用檢查檢查數據庫表是否有主鍵。
insp = inspect(engine)

# 存款利率
data = ts.get_deposit_rate()
data.to_sql(name="ts_deposit_rate", con=engine, schema=common.MYSQL_DB, if_exists='append',
            dtype={col_name: NVARCHAR(length=255) for col_name in data.columns.tolist()}, index=False)
if insp.get_primary_keys("ts_deposit_rate") == []:#判斷是否存在主鍵
    with engine.connect() as con:
        con.execute('ALTER IGNORE TABLE `ts_deposit_rate` ADD PRIMARY KEY (`date`,`deposit_type`);')

對 tushare的存款利率進行處理。

對代碼進行下封裝批量處理下 TuShare 數據：

#!/usr/local/bin/python
# -*- coding: utf-8 -*-

import sys
import time
import pandas as pd
import tushare as ts
from sqlalchemy.types import NVARCHAR
from sqlalchemy import inspect
import datetime

MYSQL_USER=""
MYSQL_PWD=""
MYSQL_HOST=""
MYSQL_DB=""

def engine():
    engine = create_engine(
        "mysql+mysqldb://" + MYSQL_USER + ":" + MYSQL_PWD + "@" + MYSQL_HOST + "/" + MYSQL_DB + "?charset=utf8",
        encoding='utf8', convert_unicode=True)
    return engine

#定義通用方法函數，插入數據庫表，並創建數據庫主鍵，保證重跑數據的時候索引唯一。
def insert_db(data, table_name, primary_keys):
    # 定義engine
    engine = engine()
    # 使用 https://docs.sqlalchemy.org/en/latest/core/reflection.html
    # 使用檢查檢查數據庫表是否有主鍵。
    insp = inspect(engine)
    data.to_sql(name=table_name, con=engine, schema=common.MYSQL_DB, if_exists='append',
                dtype={col_name: NVARCHAR(length=255) for col_name in data.columns.tolist()}, index=False)
    # 判斷是否存在主鍵
    if insp.get_primary_keys(table_name) == []:
        with engine.connect() as con:
            # 執行數據庫插入數據。
            con.execute('ALTER IGNORE TABLE `%s` ADD PRIMARY KEY (%s);' % (table_name, primary_keys))


####### 3.pdf 方法。宏觀經濟數據
def stat_all(tmp_datetime):
    # 存款利率
    data = ts.get_deposit_rate()
    insert_db(data, "ts_deposit_rate", "`date`,`deposit_type`")

    # 貸款利率
    data = ts.get_loan_rate()
    insert_db(data, "ts_loan_rate", "`date`,`loan_type`")

    # 存款準備金率
    data = ts.get_rrr()
    insert_db(data, "ts_rrr", "`date`")

    # 貨幣供應量
    data = ts.get_money_supply()
    insert_db(data, "ts_money_supply", "`month`")

    # 貨幣供應量(年底餘額)
    data = ts.get_money_supply_bal()
    insert_db(data, "ts_money_supply_bal", "`year`")

    # 國內生產總值(年度)
    data = ts.get_gdp_year()
    insert_db(data, "ts_gdp_year", "`year`")

    # 國內生產總值(季度)
    data = ts.get_gdp_quarter()
    insert_db(data, "ts_get_gdp_quarter", "`quarter`")

    # 三大需求對GDP貢獻
    data = ts.get_gdp_for()
    insert_db(data, "ts_gdp_for", "`year`")

    # 三大產業對GDP拉動
    data = ts.get_gdp_pull()
    insert_db(data, "ts_gdp_pull", "`year`")

    # 三大產業貢獻率
    data = ts.get_gdp_contrib()
    insert_db(data, "ts_gdp_contrib", "`year`")

    # 居民消費價格指數
    data = ts.get_cpi()
    insert_db(data, "ts_cpi", "`month`")

    # 工業品出廠價格指數
    data = ts.get_ppi()
    insert_db(data, "ts_ppi", "`month`")


# main函數入口
if __name__ == '__main__':
    # 使用方法傳遞。
    stat_all()

增加後的數據庫表：

存儲的數據：使用TuShare 入庫的工業品出廠價格指數

4，總結

本文的原文連接是: https://blog.csdn.net/freewebsys/article/details/75669782 未經博主允許不得轉載。
博主地址是：https://blog.csdn.net/freewebsys

TuShare 和pandas 都是非常強大的。配合起來非常的方便，抓取後的股票數據直接入庫。
在這個基礎上進行優化，把字段修改成varchar 類型，並且增加索引主鍵。

同時利用 sqlalchemy 對數據索引進行檢查如果沒有再創建。
完成了對數據的增量更新。同時保證數據不重複。