大家好，我是 Jack。
之前一直有小伙伴问我，有没有免费的股票信息查询的 API 接口？
我看了一圈，很多免费的 API 接口都年久失修，失效了。
那好吧，咱自己写一个。
想要玩量化交易，第一步，那得有稳定的股票数据来源。
然后再谈什么量化策略，怎么选股、何时买股。
怎么稳定的获取数据呢？
只能是抄起我的老板行，写个网络爬虫，自动抓取数据。
玩股票、玩基金的，应该多多少听过一款股票交流 APP 雪球。
这里面的数据很全，就它了！
前方提醒：使用网络爬虫，请控制好访问频率。
在雪球上，想要获得各种股票信息，那需要携带身份信息，也就是要有 Cookie。
没有 Cookie，很多信息是获取不到的。
2017 年的时候，我就写过关于 Cookie 的文章。
一些基础知识忘记的小伙伴，可以重温下我这个系列的文章。
模拟登录 - 准备篇
登录之后，我们就可以用保存身份信息的 Cookie，获取我们想要的各种数据：股票信息、基金信息等。
手动登录
第一步：点击登录按钮。
第二步：输入帐号和密码，并点击登录。
第三步：解锁滑块。
第四步：登录成功。
模拟登录
这里我使用 Selenium，它是一款自动化测试工具。
不过说实话，Selenium 这东西挺老了。
现在有不少更好的工具，不过对于模拟登录的知识储备，我还停留在 2017 年，也只会用它了。
有更好更好的方法的话，欢迎小伙伴们提交 PR。
不过，好在 Selenium 虽然老了点，但还能勉强胜任获取 Cookie 这项工作的。
Selenium 不会的小伙伴，可以看我从前的教程：
https://jackcui.blog.csdn.net/article/details/72331737
想要使用 Selenium，首先需要下载浏览器驱动，这里以 Chrome 浏览器为例。
打开 Chrome 浏览器，查看 Chrome 版本号。
然后根据这个版本号，下载相同大版本的驱动。
http://chromedriver.storage.googleapis.com/index.html
根据自己的操作系统，选择对应的版本。
我的是 Windows 电脑，选择 Win32 的版本。
下载好后，解压备用。
最后安装 Selenium 第三方依赖库。
python -m pip install selenium==3.4 --user
注意，需要安装 3.4 的版本，Selenium 的新版本改动较多，用我的代码会存在接口不兼容的情况。
模拟登录 - 实战篇
https://cuijiahua.com/wp-content/uploads/2022/12/1.mp4
（PS：录屏时间 12.2，由于大家都知道的原因，页面为黑白）
其实模拟登录的思路很简单，就是根据审查元素，找到各个元素的位置。
比如登录按钮，右键审查元素，然后选择 copy Xpath。
就能拷贝路径地址。
使用这种方法，找到帐号输入框、密码输入框的位置，然后点击登录即可。
这里的难点在于验证码。
不过好在，GEETEST 验证码的破解，我还是有些经验的，17 年的时候，就写过相关内容。
很多代码，直接复用即可。
整体思路就是：
直接放代码：
from selenium import webdriver
from selenium.webdriver import ActionChains
from io import BytesIO
import json
import base64
import time
from selenium.webdriver.common.by import By
from selenium.webdriver.support.wait import webdriverwait
from selenium.webdriver.support import expected_conditions as EC
from time import sleep
from PIL import Image
from selenium import webdriver
# 账号
USERNAME = '***'
# 密码
PASSWORD = '***'
BORDER = 6
class Login(object):
def __init__(self):
self.url = 'https://xueqiu.com/'
opt = webdriver.ChromeOptions()
opt.add_experimental_option('w3c', False)
self.browser = webdriver.Chrome("chromedriver.exe", chrome_options=opt)
self.browser.maximize_window()#第一处修复，设置浏览器全屏
self.username = USERNAME
self.password = PASSWORD
self.wait = webdriverwait(self.browser, 20)
def __del__(self):
print("close")
def open(self):
self.browser.get(self.url)
ele = self.browser.find_element_by_xpath('//*[@id="app"]/nav/div[1]/div[2]/div/div')#第二处修复，改xpath
ele.click()
username = self.wait.until(EC.presence_of_element_located((By.XPATH, '//input[@name="username"]')))
pwd = self.wait.until(EC.presence_of_element_located((By.XPATH, '//input[@name="password"]')))
username.send_keys(self.username)
time.sleep(2)
pwd.send_keys(self.password)
# 获取验证码按钮
def get_yzm_button(self):
button = self.wait.until(EC.presence_of_element_located((By.XPATH, '/html/body/div[2]/div[1]/div/div/div/div[2]/div[2]/div[2]')))#第三处修复，改xpath
return button
# 获取验证码图片对象
def get_img_element(self):
element = self.wait.until(EC.presence_of_element_located((By.XPATH, '//cavas[@name="geetest_canvas_bg geetest_absolute"]')))
return element
def get_position(self):
# 获取验证码位置
element = self.get_img_element()
sleep(2)
location = element.location
size = element.size
top, bottom, left, right = location['y'], location['y'] + size['height'], location['x'], location['x'] + size[
'width']
return left, top, right, bottom
def get_geetest_image(self):
"""
获取验证码图片
:return: 图片对象
"""
'''
<canvas class="geetest_canvas_bg geetest_absolute" height="160" width="260"></canvas>
'''
# 带阴影的图片
# im = self.wait.until(EC.presence_of_element_located((By.XPATH, '/html/body/div[4]/div[2]/div[6]/div/div[1]/div[1]/div/a/div[1]/div/canvas[1]')))
im = self.wait.until(EC.presence_of_element_located((By.CSS_SELECTOR, '.geetest_canvas_bg')))
time.sleep(2)
im.screenshot('captcha.png')
# 执行 JS 代码并拿到图片 base64 数据
JS = 'return document.getElementsByClassName("geetest_canvas_fullbg")[0].toDataURL("image/png");' # 不带阴影的完整图片
im_info = self.browser.execute_script(JS) # 执行js文件得到带图片信息的图片数据
# 拿到base64编码的图片信息
im_base64 = im_info.split(',')[1]
# 转为bytes类型
captcha1 = base64.b64decode(im_base64)
# 将图片保存在本地
with open('captcha1.png', 'wb') as f:
f.write(captcha1)
JS = 'return document.getElementsByClassName("geetest_canvas_bg")[0].toDataURL("image/png");'
# 执行 JS 代码并拿到图片 base64 数据ng # 带阴影的图片
im_info = self.browser.execute_script(JS) # 执行js文件得到带图片信息的图片数据
# 拿到base64编码的图片信息
im_base64 = im_info.split(',')[1]
# 转为bytes类型
captcha2 = base64.b64decode(im_base64)
# 将图片保存在本地
with open('captcha2.png', 'wb') as f:
f.write(captcha2)
captcha1 = Image.open('captcha1.png')
captcha2 = Image.open('captcha2.png')
return captcha1, captcha2
# 获取网页截图
def get_screen_shot(self):
screen_shot = self.browser.get_screenshot_as_png()
screen_shot = Image.open(BytesIO(screen_shot))
return screen_shot
def get_yzm_img(self, name='captcha.png'):
# 获取验证码图片
left, bottom = self.get_position()
print('验证码位置', right)
screen_shot = self.get_screen_shot()
captcha = screen_shot.crop((left, bottom))
captcha.save(name)
return captcha
def get_slider(self):
# 获取滑块
# :return: 滑块对象
slider = self.wait.until(EC.element_to_be_clickable((By.CLASS_NAME, 'geetest_slider_track')))
return slider
def get_gap(self, image1, image2):
"""
获取缺口偏移量
:param image1: 不带缺口图片
:param image2: 带缺口图片
:return:
"""
left = 62
for i in range(left, image1.size[0]):
for j in range(image1.size[1]):
if not self.is_pixel_equal(image1, image2, i, j):
left = i
return left
# return left
def is_pixel_equal(self, x, y):
"""
判断两个像素是否相同
:param image1: 图片1
:param image2: 图片2
:param x: 位置x
:param y: 位置y
:return: 像素是否相同
"""
# 取两个图片的像素点
pixel1 = image1.load()[x, y]
pixel2 = image2.load()[x, y]
threshold = 60
if abs(pixel1[0] - pixel2[0]) < threshold and abs(pixel1[1] - pixel2[1]) < threshold and abs(
pixel1[2] - pixel2[2]) < threshold:
return True
else:
return False
def get_track(self, distance):
"""
根据偏移量获取移动轨迹
:param distance: 偏移量
:return: 移动轨迹
"""
# 初速度
v = 0
# 单位时间为0.2s来统计轨迹，轨迹即0.2内的位移
t = 0.3
# 位移/轨迹列表，列表内的一个元素代表0.2s的位移
tracks = []
# 当前的位移
current = 5
# 到达mid值开始减速
mid = distance * 3 / 5
while current < distance:
if current < mid:
# 加速度越小，单位时间的位移越小,模拟的轨迹就越多越详细
a = 2
else:
a = -3
# 初速度
v0 = v
# 0.2秒时间内的位移
s = v0 * t + 0.4 * a * (t ** 2)
# 当前的位置
current += s
# 添加到轨迹列表
tracks.append(round(s))
# 速度已经达到v,该速度作为下次的初速度
v = v0 + a * t
return tracks
def move_to_gap(self, slider, track):
"""
拖动滑块到缺口处
:param slider: 滑块
:param track: 轨迹
:return:
"""
ActionChains(self.browser).click_and_hold(slider).perform()
for x in track:
ActionChains(self.browser).move_by_offset(xoffset=x, yoffset=0).perform()
time.sleep(0.5)
ActionChains(self.browser).release().perform()
def shake_mouse(self):
"""
模拟人手释放鼠标抖动
:return: None
"""
ActionChains(self.browser).move_by_offset(xoffset=-2, yoffset=0).perform()
ActionChains(self.browser).move_by_offset(xoffset=2, yoffset=0).perform()
def operate_slider(self, track):
'''
拖动滑块
'''
# 获取拖动按钮
back_tracks = [-1,-1, -1, -1]
slider_bt = self.browser.find_element_by_xpath('//div[@class="geetest_slider_button"]')
# 点击拖动验证码的按钮不放
ActionChains(self.browser).click_and_hold(slider_bt).perform()
# 按正向轨迹移动
for i in track:
ActionChains(self.browser).move_by_offset(xoffset=i, yoffset=0).perform()
time.sleep(1)
ActionChains(self.browser).release().perform()
def get_cookies(self):
try:
cookie_list = self.browser.get_cookies()
cookie_dict = {i['name']: i['value'] for i in cookie_list}
with open('xueqiu_cookies', 'w', encoding='utf8')as f:
cookie_dict = json.dumps(cookie_dict)
f.write(cookie_dict)
return cookie_dict
except:
print("cookie 获取失败")
return None
# 读取cookie
def return_cookie(self):
cookies = ''
with open('xueqiu_cookies', 'r')as f:
cookie = f.read()[1:-1]
cookie = cookie.split(', ')
for i in cookie:
cook = i.split(': ')
cookies += cook[0][1:-1] + '=' + cook[1][1:-1] + ';'
return cookies
def run(self):
# 破解入口
self.open(), sleep(3)
self.get_yzm_button().click(), sleep(2)# 点击验证按钮
# 点按呼出缺口
slider = self.get_slider()
# slider.click()
# 获取带缺口的验证码图片
image1, image2 = self.get_geetest_image()
gap = self.get_gap(image1, image2)
print('缺口位置', gap)
track = self.get_track(gap)
print('滑动轨迹', track)
self.operate_slider(track)
# 判定是否成功
time.sleep(8)
try:
elem = self.wait.until(
EC.text_to_be_present_in_element((By.CLASS_NAME, 'nav__btn--longtext'), '发帖'))
if elem:
cookie = self.get_cookies()
else:
print("get cookies errors")
except Exception as e:
print(e, 'fail! ')
time.sleep(3)
self.run()
finally:
self.browser.quit()
if __name__ == '__main__':
crack = Login()
crack.run()
代码我也上传到 Github 上了，代码的后续更新维护会放在这里，建议 Star 收藏下。
https://github.com/Jack-Cherish/quantitative
数据获取
等待模拟登录完成后，会保存一个名为 xueqiu_cookies 的文件。
这里保存的是帐号的 Cookie，使用这个 Cookie 就能获取雪球的数据了。
比如，获取一下股票实时行情和现金流量表，就可以这样写：
#-*- coding:utf-8 -*-
import requests
import json
def fetch(url, xq_a_token):
headers = {'Host': 'stock.xueqiu.com', 'Accept': 'application/json', 'Cookie': 'xq_a_token={};'.format(xq_a_token), 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/107.0.0.0 Safari/537.36', 'Accept-Language': 'zh-Hans-CN;q=1, ja-JP;q=0.9', 'Accept-Encoding': 'br, gzip, deflate', 'Connection': 'keep-alive'}
response = requests.get(url, headers = headers)
if response.status_code != 200:
raise Exception(response.content)
return json.loads(response.content)
if __name__ == '__main__':
# 获取股票 SH600000 实时行情
url = "http://stock.xueqiu.com/v5/stock/quote.json?extend=detail&symbol=SH600000&count=10"
with open("xueqiu_cookies", "r") as f:
cookies_info = json.load(f)
res = fetch(url, cookies_info['xq_a_token'])
print(res)
# 获取股票 SH600000 现金流量表
url = "http://stock.xueqiu.com/v5/stock/finance/cn/cash_flow.json?symbol=SH600000&count=10"
with open("xueqiu_cookies", cookies_info['xq_a_token'])
print(res)
运行结果：
有了 Cookie，很多接口数据都能获取，实时行情、实时分笔、业绩预告、机构评级、资金流向趋势、资金流向历史、资金成交分布、大宗交易、融资融券、业绩指标、利润表、资产负债表、现金流量表、主营业务构成、F10 十大股东、F10 主要指标等等。
这些数据，都能获取。
絮叨
篇幅有限，今天就是带大家小小实战下。
后续我会完善各个常用查询接口，方便大家获取各类数据，用于量化分析。
万事开头难，先弄好数据，再看量化策略~
如果喜欢这类的内容，记得点赞，喜欢的人多的话，我会快速加更的~
最后必须提醒一下各位：
对了，还有不少小伙伴问我，我的量化策略收益如何。
去年的五万元实验，最后浮盈不到 10%，清仓之后就换新的策略实验了。
6月份的时候，又用上了新策略，新的策略一直跑到今年 10 月份，也就这样：
实验没放多少钱，随便玩玩，你觉得，这点收益如何？
好了，今天就聊这么多吧，我是 Jack，我们下期见~
