1、安裝python,執(zhí)行以下腳本

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依賴(lài)，詳情參照文檔

案例1：獲取板塊數(shù)據(jù)

import QUANTAXIS as QA

import mysql.connector

import pandas as pd

import time

from retrying import retry

mydb = mysql.connector.connect(

host="", # 數(shù)據(jù)庫(kù)主機(jī)地址

user="", # 數(shù)據(jù)庫(kù)用戶(hù)名

passwd="", # 數(shù)據(jù)庫(kù)密碼

database="" # 數(shù)據(jù)庫(kù)

)

mycursor = mydb.cursor()

QA.QA_util_log_info('板塊數(shù)據(jù)')

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() # 數(shù)據(jù)表內(nèi)容有更新，必須使用到該語(yǔ)句

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="", # 數(shù)據(jù)庫(kù)主機(jī)地址

user="", # 數(shù)據(jù)庫(kù)用戶(hù)名

passwd="", # 數(shù)據(jù)庫(kù)密碼

database="" # 數(shù)據(jù)庫(kù)

)

mycursor = mydb.cursor()

QA.QA_util_log_info('指數(shù)列表')

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() # 數(shù)據(jù)表內(nèi)容有更新，必須使用到該語(yǔ)句

print(mycursor.rowcount, "記錄插入成功。")

案例3：獲取股票指數(shù)列表

import QUANTAXIS as QA

import mysql.connector

import pandas as pd

import time

import sys

from retrying import retry

mydb = mysql.connector.connect(

host="", # 數(shù)據(jù)庫(kù)主機(jī)地址

user="", # 數(shù)據(jù)庫(kù)用戶(hù)名

passwd="", # 數(shù)據(jù)庫(kù)密碼

database="" # 數(shù)據(jù)庫(kù)

)

mycursor = mydb.cursor()

QA.QA_util_log_info('指數(shù)列表')

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() # 數(shù)據(jù)表內(nèi)容有更新，必須使用到該語(yǔ)句

print(mycursor.rowcount, "記錄插入成功。")

案例4：獲取股票日線(xiàn)

import QUANTAXIS as QA

import mysql.connector

import pandas as pd

import time

import sys

from retrying import retry

mydb = mysql.connector.connect(

host="", # 數(shù)據(jù)庫(kù)主機(jī)地址

user="", # 數(shù)據(jù)庫(kù)用戶(hù)名

passwd="", # 數(shù)據(jù)庫(kù)密碼

database="" # 數(shù)據(jù)庫(kù)

)

mycursor = mydb.cursor()

QA.QA_util_log_info('股票日線(xiàn)')

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() # 數(shù)據(jù)表內(nèi)容有更新，必須使用到該語(yǔ)句

print("記錄插入成功。")

5、獲取指數(shù)日線(xiàn)

import QUANTAXIS as QA

import mysql.connector

import pandas as pd

import time

import sys

from retrying import retry

mydb = mysql.connector.connect(

host="", # 數(shù)據(jù)庫(kù)主機(jī)地址

user="", # 數(shù)據(jù)庫(kù)用戶(hù)名

passwd="", # 數(shù)據(jù)庫(kù)密碼

database="" # 數(shù)據(jù)庫(kù)

)

mycursor = mydb.cursor()

QA.QA_util_log_info('指數(shù)日線(xiàn)')

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() # 數(shù)據(jù)表內(nèi)容有更新，必須使用到該語(yǔ)句

print("記錄插入成功。")

6、獲取指數(shù)分鐘線(xiàn)

import QUANTAXIS as QA

import mysql.connector

import pandas as pd

import time

import sys

from retrying import retry

mydb = mysql.connector.connect(

host="", # 數(shù)據(jù)庫(kù)主機(jī)地址

user="", # 數(shù)據(jù)庫(kù)用戶(hù)名

passwd="", # 數(shù)據(jù)庫(kù)密碼

database="" # 數(shù)據(jù)庫(kù)

)

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="", # 數(shù)據(jù)庫(kù)主機(jī)地址

user="", # 數(shù)據(jù)庫(kù)用戶(hù)名

passwd="", # 數(shù)據(jù)庫(kù)密碼

database="" # 數(shù)據(jù)庫(kù)

)

mycursor = mydb.cursor()

- END -

轉(zhuǎn)載：感謝您對(duì)張偉江個(gè)人博客網(wǎng)站平臺(tái)的認(rèn)可，以及對(duì)我們?cè)瓌?chuàng)作品以及文章的青睞，非常歡迎各位朋友分享到個(gè)人站長(zhǎng)或者朋友圈，但轉(zhuǎn)載請(qǐng)說(shuō)明文章出處“來(lái)源張偉江個(gè)人博客”。

總結(jié)

以上是生活随笔為你收集整理的python调用通达信函数_python获取通达信基本数据源码的全部?jī)?nèi)容，希望文章能夠幫你解決所遇到的問(wèn)題。

如果覺(jué)得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。