15 KiB
15 KiB
In [ ]:
import baostock as bs
import pandas as pd
#### 登陆系统 ####
lg = bs.login()
# 显示登陆返回信息
print('login respond error_code:'+lg.error_code)
print('login respond error_msg:'+lg.error_msg)
#### 获取证券信息 ####
rs = bs.query_all_stock(day="2020-10-20")
print('query_all_stock respond error_code:'+rs.error_code)
print('query_all_stock respond error_msg:'+rs.error_msg)
#### 打印结果集 ####
data_list = []
while (rs.error_code == '0') & rs.next():
# 获取一条记录，将记录合并在一起
data_list.append(rs.get_row_data())
#results = pd.DataFrame(data_list, columns=rs.fields)
#### 结果集输出到csv文件 ####
#result.to_csv("all_stock.csv", encoding="utf-8", index=False)
for result in data_list:
print(result)
#### 登出系统 ####
bs.logout()In [ ]:
import baostock as bs
import pandas as pd
import pymysql
# 登陆系统
lg = bs.login()
# 显示登陆返回信息
print('login respond error_code:'+lg.error_code)
print('login respond error_msg:'+lg.error_msg)
# 获取证券基本资料
rs = bs.query_stock_basic(code="")
# rs = bs.query_stock_basic(code_name="浦发银行") # 支持模糊查询
print('query_stock_basic respond error_code:'+rs.error_code)
print('query_stock_basic respond error_msg:'+rs.error_msg)
# 打印结果集
data_list = []
while (rs.error_code == '0') & rs.next():
# 获取一条记录，将记录合并在一起
data_list.append(rs.get_row_data())
#result = pd.DataFrame(data_list, columns=rs.fields)
# 结果集输出到csv文件
#result.to_csv("D:/stock_basic.csv", encoding="gbk", index=False)
db = pymysql.connect("localhost","root","songyi","stock" )
cursor = db.cursor()
sql = "insert into stock_info (code,name,ipoDate,outDate,type,status) values(%s,%s,%s,%s,%s,%s)"
try:
cursor.executemany(sql,tuple(data_list))
db.commit()
print("ok!")
except:
# 如果发生错误则回滚
print("error!")
db.rollback()
db.close()
# 登出系统
bs.logout()In [ ]:
import csv
import pymysql
def read_data(filename):
detail = {}
with open(filename) as f:
reader = csv.reader(f)
#header_row =next(reader)
for row in reader:
detail.setdefault(row[0],[])
detail[row[0]].append(row[1])
return detail
m_xx = []
m_stock = {}
db = pymysql.connect("localhost","root","songyi","stock" )
cursor = db.cursor()
sql = 'select code,name from stock_info where type="1"'
cursor.execute(sql)
results = cursor.fetchall()
for result in results:
m_stock[result[0]] = result[1]
#print(m_stock)
filename = '每日标的信息.csv'
detail = read_data(filename)
for k,v in detail.items():
m_rq = '2020-' + k[0:2] + '-' + k[2:]
i = 1
for m_dm in v:
if m_dm[0:1] == '6':
m_dm = 'sh.' + m_dm
else:
m_dm = 'sz.' + m_dm
print('\t'+m_dm + '\t'+m_stock[m_dm])
m_xx.append((m_rq,m_dm,i))
i += 1
choice = input('以上为本日数据，是否导入？（y/n）')
if choice.upper() == "Y":
sql = "insert into daily_item (rq,code,ord) values(%s,%s,%s)"
try:
cursor.executemany(sql,m_xx)
db.commit()
print("ok!")
except:
# 如果发生错误则回滚
print("error!")
db.rollback()
db.close()In [ ]:
import csv
import pymysql
def read_data(filename):
detail = {}
with open(filename) as f:
reader = csv.reader(f)
# header_row =next(reader)
for row in reader:
detail1 = {}
detail.setdefault(row[3],[])
detail1 = {'dm':row[0],'zj':row[1],'ly':row[2],'cb':row[4]}
detail[row[3]].append(detail1)
return detail
m_xx = []
m_add = []
m_sub = []
db = pymysql.connect("localhost","root","songyi","stock" )
cursor = db.cursor()
filename = '调仓明细.csv'
detail = read_data(filename)
#print(detail)
for rq in sorted(detail.keys()):
m_rq = '2020-' + rq[0:2] + '-' + rq[2:]
for xx_move in detail[rq]:
m_dm = xx_move['dm']
if m_dm[0:1] == '6':
m_dm = 'sh.' + m_dm
else:
m_dm = 'sz.' + m_dm
m_xx.append((m_rq,m_dm,int(xx_move['zj']),xx_move['ly']))
if xx_move['zj'] == '1':
m_add.append((m_dm,xx_move['ly'],float(xx_move['cb'])))
else:
m_sub.append((m_dm))
#print(m_xx)
sql = "insert into change_item (rq,code,pos,reason) values(%s,%s,%s,%s)"
try:
cursor.executemany(sql,m_xx)
if len(m_add) > 0:
sql_add = "insert into stock_item (code,reason,cost) values(%s,%s,%s)"
cursor.executemany(sql_add,m_add)
if len(m_sub) > 0:
sql_sub = "update stock_item set status=0 where code=%s"
cursor.executemany(sql_sub,m_sub)
db.commit()
print("导入成功!")
except:
# 如果发生错误则回滚
print("error!")
db.rollback()
db.close()In [ ]:
import pymysql
db = pymysql.connect("localhost","root","songyi","stock" )
cursor = db.cursor()
sql = 'SELECT a.unit,a.CODE,b.cost FROM daily_item AS a,daily_cost as b WHERE a.rq=(select max(rq) FROM daily_item) AND a.code=b.code'
cursor.execute(sql)
results = cursor.fetchall()
sql = "insert into stock_item(item,code,cost) values(%s,%s,%s)"
try:
cursor.executemany(sql,list(results))
#db.commit()
print("ok!")
except:
# 如果发生错误则回滚
print("error!")
db.rollback()
sql = 'SELECT a.reason,a.code from change_item AS a WHERE a.pos=1'
cursor.execute(sql)
results = cursor.fetchall()
sql = "update stock_item set reason=%s where code=%s "
try:
cursor.executemany(sql,list(results))
#db.commit()
print("ok!")
except:
# 如果发生错误则回滚
print("error!")
db.rollback()
db.close()In [ ]:
import requests
import time
from bs4 import BeautifulSoup
import pymongo
import re
#pattern = re.compile('(?<=\").*(?=\")')
pattern = re.compile(r'\"(.*)\"')
#myclient = pymongo.MongoClient('mongodb://localhost:27017/')
#mydb = myclient["gaokao"]
#mycol = mydb["news"]
url = 'http://hq.sinajs.cn/list=USDCAD'
strhtml = requests.get(url)
#strhtml.encoding = 'utf8'
soup = BeautifulSoup(strhtml.text,'lxml')
data = strhtml.text
#data1 = data.split("\r")
#data = soup.select('schoolList')
#for item1 in data:
# print(item1.get_text())
if pattern.findall(data):
for data1 in pattern.findall(data):
data2 = data1.split(',')
print(data2)
print('当前买入价：',data2[1])
print('当前卖出价：',data2[2])
print('昨收价：',data2[3])
print('今开价：',data2[5])
print('最高价：',data2[6])
print('最低价：',data2[7])
print(float(data2[7]))In [ ]:
In [ ]:
from email.mime.text import MIMEText
from email.header import Header
import smtplib
import requests
import time
import re
pattern = re.compile(r'\"(.*)\"')
url = 'http://hq.sinajs.cn/list=USDCAD'
strhtml = requests.get(url)
data = strhtml.text
if pattern.findall(data):
for data1 in pattern.findall(data):
data2 = data1.split(',')
#print(data2)
with open('price.txt','r') as fl:
for line in fi:
p_high = line.split(',')[0]
p_low = line.split(',')[1]
message ='当前美元加元买入价：{}，卖出价：{}'.format(data2[1],data2[2])
msg = MIMEText(message,'plain','utf-8')
msg['Subject'] = Header("外汇价格已经到达预期价位！",'utf-8')
msg['From'] = Header('512song@sina.com')
msg['To'] = Header('491525765@qq.com','utf-8')
from_addr = '512song@sina.com' #发件邮箱
password = '409fe5d8471da663' #邮箱密码
to_addr = 'songyi@yeah.net' #收件邮箱
smtp_server = 'smtp.sina.com' #SMTP服务器，以新浪为例
try:
server = smtplib.SMTP(smtp_server,25) #第二个参数为默认端口为25，有些邮件有特殊端口
print('开始登录')
server.set_debuglevel(1)
server.login(from_addr,password) #登录邮箱
print('登录成功')
print("邮件开始发送")
server.sendmail(from_addr,to_addr,msg.as_string()) #将msg转化成string发出
server.quit()
print("邮件发送成功")
except smtplib.SMTPException as e:
print("邮件发送失败",e)
In [ ]:
from email.mime.text import MIMEText
from email.header import Header
import smtplib
import requests
import time
import re
pattern = re.compile(r'\"(.*)\"')
url = 'http://hq.sinajs.cn/list=USDCAD'
strhtml = requests.get(url)
data = strhtml.text
if pattern.findall(data):
for data1 in pattern.findall(data):
data2 = data1.split(',')
print(data2)
message ='当前美元加元最高价：{}，最低价：{}'.formtat()
msg = MIMEText(message,'plain','utf-8')
msg['Subject'] = Header("测试smtp邮件",'utf-8')
msg['From'] = Header('512song@sina.com')
msg['To'] = Header('491525765@qq.com','utf-8')
from_addr = '512song@sina.com' #发件邮箱
password = '409fe5d8471da663' #邮箱密码
to_addr = '491525765@qq.com' #收件邮箱
smtp_server = 'smtp.sina.com' #SMTP服务器，以新浪为例
try:
server = smtplib.SMTP(smtp_server,25) #第二个参数为默认端口为25，有些邮件有特殊端口
print('开始登录')
server.set_debuglevel(1)
server.login(from_addr,password) #登录邮箱
print('登录成功')
print("邮件开始发送")
server.sendmail(from_addr,to_addr,msg.as_string()) #将msg转化成string发出
server.quit()
print("邮件发送成功")
except smtplib.SMTPException as e:
print("邮件发送失败",e)
In [ ]:
import pygal
bar_chart = pygal.Bar(height=300)
bar_chart.add('Fibonacci', [0, 1, 1, 2, 3, 5, 8, 13, 21, 34, 55])
bar_chart.add('Padovan', [1, 1, 1, 2, 2, 3, 4, 5, 7, 9, 12])
svg = bar_chart.render()In [ ]:
from IPython.display import SVG
SVG(svg)In [ ]: