python获取通达信基本数据源码

1、安装python,执行以下脚本

wget https://repo.anaconda.com/miniconda/Miniconda3-latest-Linux-x86_64.sh
chmod a+x Miniconda3-latest-Linux-x86_64.sh
./Miniconda3-latest-Linux-x86_64.sh

2、python安装quanaxis依赖，详情参照文档

案例1：获取板块数据

import QUANTAXIS as QA
import mysql.connector
import pandas as pd
import time
from retrying import retry
mydb = mysql.connector.connect(
  host="",       # 数据库主机地址
  user="",    # 数据库用户名
  passwd="",   # 数据库密码
  database=""  # 数据库
)
mycursor = mydb.cursor() 

QA.QA_util_log_info('板块数据')
data=QA.QAFetch.QATdx.QA_fetch_get_stock_block()
data = pd.DataFrame(data.values,columns=data.columns)
sql = "INSERT INTO fa_stock_block (blockname, code, type, source, createtime, updatetime) VALUES (%s, %s, %s, %s,%s,%s)"
for i in range(len(data.index)):
    intime = int(time.time())
    val = (data['blockname'][i],data['code'][i],data['type'][i],data['source'][i],intime,intime)
    mycursor.execute(sql, val)
    print(val)
mydb.commit()    # 数据表内容有更新，必须使用到该语句
print(mycursor.rowcount, "记录插入成功。")

import QUANTAXIS as QA
import mysql.connector
import pandas as pd
import time
from retrying import retry
mydb = mysql.connector.connect(
  host="",       # 数据库主机地址
  user="",    # 数据库用户名
  passwd="",   # 数据库密码
  database=""  # 数据库
)
mycursor = mydb.cursor() 
QA.QA_util_log_info('指数列表')
data=QA.QAFetch.QATdx.QA_fetch_get_stock_list('index')
data = pd.DataFrame(data.values,columns=data.columns)
sql = "INSERT INTO fa_stock_list (code, name, sse, sec, volunit,decimal_point,pre_close,createtime, updatetime) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s)"
for i in range(len(data.index)):
    intime = int(time.time())
    val = (data['code'][i],data['name'][i],data['sse'][i],data['sec'][i],data['volunit'][i],data['decimal_point'][i],data['pre_close'][i],intime,intime)
    print(val)
    mycursor.execute(sql, val)
mydb.commit()    # 数据表内容有更新，必须使用到该语句
print(mycursor.rowcount, "记录插入成功。")

案例3：获取股票指数列表

import QUANTAXIS as QA
import mysql.connector
import pandas as pd
import time
import sys
from retrying import retry
mydb = mysql.connector.connect(
  host="",       # 数据库主机地址
  user="",    # 数据库用户名
  passwd="",   # 数据库密码
  database=""  # 数据库
)
mycursor = mydb.cursor() 

QA.QA_util_log_info('指数列表')
data=QA.QAFetch.QATdx.QA_fetch_get_stock_list('index')
data = pd.DataFrame(data.values,columns=data.columns)
sql = "INSERT INTO fa_stock_list (code, name, sse, sec, volunit,decimal_point,pre_close,createtime, updatetime) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s)"
for i in range(len(data.index)):
    intime = int(time.time())
    val = (data['code'][i],data['name'][i],data['sse'][i],data['sec'][i],data['volunit'][i],data['decimal_point'][i],data['pre_close'][i],intime,intime)
    print(val)
    mycursor.execute(sql, val)
mydb.commit()    # 数据表内容有更新，必须使用到该语句
print(mycursor.rowcount, "记录插入成功。")

import QUANTAXIS as QA
import mysql.connector
import pandas as pd
import time
import sys
from retrying import retry
mydb = mysql.connector.connect(
  host="",       # 数据库主机地址
  user="",    # 数据库用户名
  passwd="",   # 数据库密码
  database=""  # 数据库
)
mycursor = mydb.cursor() 

QA.QA_util_log_info('股票日线')
stock_list=QA.QAFetch.QATdx.QA_fetch_get_stock_list('stock')
stock_list = pd.DataFrame(stock_list.values,columns=stock_list.columns)
for ii in range(len(stock_list.index)):
    @retry(stop_max_attempt_number=5, wait_random_min=100, wait_random_max=10000)
    def get_index_day(code,start,end):
        #print(stock_list['code'][ii])
        data=QA.QAFetch.QATdx.QA_fetch_get_stock_day(code,start,end)
        return data 
    try:
        data = get_index_day(stock_list['code'][ii],'2000-01-01','2019-08-28')       
        data = pd.DataFrame(data.values,columns=data.columns)
        insert_list = []
        sql = "INSERT INTO fa_stock_day (code, open, close, high, low,vol,amount,date,date_stamp,createtime, updatetime) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)"
        for i in range(len(data.index)):
            intime = int(time.time())
            val = (data['code'][i],data['open'][i],data['close'][i],data['high'][i],data['low'][i],data['vol'][i],data['amount'][i],data['date'][i],data['date_stamp'][i],intime,intime)
            insert_list.append(val)
        mycursor.executemany(sql, insert_list)
        #print(mycursor.rowcount, "记录插入成功。",stock_list['code'][ii])
    except BaseException as e:
        print(e)
        continue
mydb.commit()    # 数据表内容有更新，必须使用到该语句 
print("记录插入成功。")

import QUANTAXIS as QA
import mysql.connector
import pandas as pd
import time
import sys
from retrying import retry
mydb = mysql.connector.connect(
  host="",       # 数据库主机地址
  user="",    # 数据库用户名
  passwd="",   # 数据库密码
  database=""  # 数据库
)
mycursor = mydb.cursor() 

QA.QA_util_log_info('指数日线')
stock_list=QA.QAFetch.QATdx.QA_fetch_get_stock_list('index')
stock_list = pd.DataFrame(stock_list.values,columns=stock_list.columns)
for ii in range(len(stock_list.index)):
    @retry(stop_max_attempt_number=5, wait_random_min=100, wait_random_max=10000)
    def get_index_day(code,start,end):
        print(stock_list['code'][ii])
        data=QA.QAFetch.QATdx.QA_fetch_get_index_day(code,start,end)
        return data 
    try:
        data = get_index_day(stock_list['code'][ii],'2000-01-01','2019-08-28')       
        data = pd.DataFrame(data.values,columns=data.columns)
        insert_list = []
        sql = "INSERT INTO fa_stock_index_day (code, open, close, high, low,vol,amount,up_count,down_count,date,date_stamp,createtime, updatetime) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)"
        for i in range(len(data.index)):
            intime = int(time.time())
            val = (data['code'][i],data['open'][i],data['close'][i],data['high'][i],data['low'][i],data['vol'][i],data['amount'][i],data['up_count'][i],data['down_count'][i],data['date'][i],data['date_stamp'][i],intime,intime)
            insert_list.append(val)
        mycursor.executemany(sql, insert_list)
        #print(mycursor.rowcount, "记录插入成功。",stock_list['code'][ii])
    except BaseException as e:
        print(e)
        continue
mydb.commit()    # 数据表内容有更新，必须使用到该语句 
print("记录插入成功。")

import QUANTAXIS as QA
import mysql.connector
import pandas as pd
import time
import sys
from retrying import retry
mydb = mysql.connector.connect(
  host="",       # 数据库主机地址
  user="",    # 数据库用户名
  passwd="",   # 数据库密码
  database=""  # 数据库
)
mycursor = mydb.cursor() 

import QUANTAXIS as QA
import mysql.connector
import pandas as pd
import time
import sys
from retrying import retry
mydb = mysql.connector.connect(
  host="",       # 数据库主机地址
  user="",    # 数据库用户名
  passwd="",   # 数据库密码
  database=""  # 数据库
)
mycursor = mydb.cursor() 

- END -