大家好,我是 Jack。
之前一直有小伙伴问我,有没有免费的股票信息查询的 API 接口?
我看了一圈,很多免费的 API 接口都年久失修,失效了。
那好吧,咱自己写一个
想要玩量化交易,第一步,那得有稳定的股票数据来源。
然后再谈什么量化策略,怎么选股、何时买股。
怎么稳定、免费地获取数据呢?
只能是抄起我的老本行,写个网络爬虫,自动抓取数据
玩股票、玩基金的,应该多听过一款股票交流 APP 雪球。
这里面的数据很全,就它了!
前方提醒:使用网络爬虫,请控制好访问频率。
在雪球上,想要获得各种股票信息,那需要携带身份信息,也就是要有 Cookie。
没有 Cookie,很多信息是获取不到的。
2017 年的时候,我就写过关于 Cookie 的文章。
一些基础知识忘记的小伙伴,可以重温下我这个系列的文章。
公众号后台-回忆录-网络爬虫:
想要获取 Cookie,那就需要进行模拟登录。

1模拟登录 - 准备篇

模拟登录,顾名思义,就是模拟人类的行为,登录这个网站。
登录之后,我们就可以用保存身份信息的 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 的新版本改动较多,用我的代码会存在接口不兼容的情况。

2模拟登录 - 实战篇

我们先睹为快,看下让代码自动登录雪球的效果:
(PS:录屏时间 12.2,由于大家都知道的原因,页面为黑白)
其实模拟登录的思路很简单,就是根据审查元素,找到各个元素的位置。
比如登录按钮,右键审查元素,然后选择 Copy Xpath。
就能拷贝路径地址。
使用这种方法,找到帐号输入框、密码输入框的位置,然后点击登录即可。
这里的难点在于验证码。
不过好在,GEETEST 验证码的破解,我还是有些经验的,17 年的时候,我也写过。
很多代码,直接复用即可
整体思路就是:
  • 使用Selenium打开页面。
  • 匹配到输入框,输入账号密码,点击登录。
  • 读取验证码图片,并做缺口识别。
  • 根据缺口位置,计算滑动距离。
  • 根据滑动距离,拖拽滑块到需要匹配的位置。
直接放代码:
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


classLogin(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"
)


defopen(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)


# 获取验证码按钮
defget_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


# 获取验证码图片对象
defget_img_element(self):
        element = self.wait.until(EC.presence_of_element_located((By.XPATH, 
'//cavas[@name="geetest_canvas_bg geetest_absolute"]'
)))

return
 element


defget_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


defget_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


# 获取网页截图
defget_screen_shot(self):
        screen_shot = self.browser.get_screenshot_as_png()

        screen_shot = Image.open(BytesIO(screen_shot))

return
 screen_shot


defget_yzm_img(self, name='captcha.png'):
#  获取验证码图片
        left, top, right, bottom = self.get_position()

        print(
'验证码位置'
, top, bottom, left, right)

        screen_shot = self.get_screen_shot()

        captcha = screen_shot.crop((left, top, right, bottom))

        captcha.save(name)

return
 captcha


defget_slider(self):
# 获取滑块
# :return: 滑块对象
        slider = self.wait.until(EC.element_to_be_clickable((By.CLASS_NAME, 
'geetest_slider_track'
)))

return
 slider


defget_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
]):

ifnot
 self.is_pixel_equal(image1, image2, i, j):

                    left = i

return
 left

# return left

defis_pixel_equal(self, image1, image2, 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:

returnTrue
else
:

returnFalse

defget_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


defmove_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()


defshake_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()


defoperate_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()


defget_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 获取失败"
)

returnNone

# 读取cookie
defreturn_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


defrun(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

3数据获取

等待模拟登录完成后,会保存一个名为 xueqiu_cookies 的文件。
这里保存的是帐号的 Cookie,使用这个 Cookie 就能获取雪球的数据了。
比如,获取一下股票实时行情和现金流量表,就可以这样写:
#-*- coding:utf-8 -*-
import
 requests

import
 json


deffetch(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"
"r"
as
 f:

        cookies_info = json.load(f)

    res = fetch(url, cookies_info[
'xq_a_token'
])

    print(res)

运行结果:
有了 Cookie,很多接口数据都能获取,实时行情、实时分笔、业绩预告、机构评级、资金流向趋势、资金流向历史、资金成交分布、大宗交易、融资融券、业绩指标、利润表、资产负债表、现金流量表、主营业务构成、F10 十大股东、F10 主要指标等等。
这些数据,都能获取。

4絮叨

篇幅有限,今天就是带大家小小实战下。
后续我会完善各个常用查询接口,方便大家获取各类数据,用于量化分析。
万事开头难,先弄好数据,再看量化策略~
如果喜欢这类的内容,记得点赞,喜欢的人多的话,我会快速加更的~
最后必须提醒一下各位:
获取数据,请温柔,请勿高并发获取,且用且珍惜。
对了,还有不少小伙伴问我,我的量化策略收益如何。
去年的五万元实验,最后浮盈不到 10%,清仓之后就换新的策略实验了。
6月份的时候,又用上了新策略,新的策略一直跑到今年 10 月份,实验账户收益:
实验没放多少钱,随便玩玩,你觉得,这点收益如何?
好了,今天就聊这么多吧,我是 Jack,我们下期见~
·················END·················
继续阅读
阅读原文