600字范文 > python调用通达信函数_python获取通达信基本数据源码

python调用通达信函数_python获取通达信基本数据源码

时间：2024-03-30 08:52:34

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

wget /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)

mit() # 数据表内容有更新，必须使用到该语句

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)

mit() # 数据表内容有更新，必须使用到该语句

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)

mit() # 数据表内容有更新，必须使用到该语句

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','-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

mit() # 数据表内容有更新，必须使用到该语句

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','-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

mit() # 数据表内容有更新，必须使用到该语句

print("记录插入成功。")

6、获取指数分钟线

import QUANTAXIS as QA

import mysql.connector

import pandas as pd