您现在的位置是: 首页> 学无止境> Python> Python

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

张伟江2019-08-29 19:23Python6741人已围观

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, "记录插入成功。")
案例2:获取股票列表

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, "记录插入成功。")
案例4:获取股票日线

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("记录插入成功。")
5、获取指数日线

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("记录插入成功。")
6、获取指数分钟线

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 -

点赞(1) 打赏

文章评论 共有 0 条评论

暂无评论

本栏推荐

猜你喜欢

站点信息

  • 建站时间:2018年10月24日
  • 网站程序:fastadmin
  • 文章统计296篇文章
  • 标签管理标签云
  • 统计数据百度统计
  • 微信号:zwj982215226

打赏本站

  • 如果你觉得本站很棒,可以通过扫码支付打赏哦!
  • 微信扫码:你说多少就多少~
  • 支付宝扫码:你说多少就多少~

微信小程序

微信扫一扫体验

立即
投稿

微信公众账号

微信扫一扫加关注

发表
评论
返回
顶部