升级 Flask 到 Quart, 3 倍性能提升就这么简单!

简评:将你的 Flask 应用程序升级到 Quart 应用程序,轻松获得 3 倍的性能提升。

自从 Flask 在 8 年前发布以来,Python 发生了很大变化,特别是引入了 asyncio之后。asyncio 允许开发像 uvloop 和 asyncpg 这样的库来大大提高性能,可惜要 Flask 集成 asyncio 或这些库并不是一件简单的事情。不过不用气馁,Quart 框架与 asyncio 可以一起使用 Flask-API。

基于共享的 Flask-API,现有的 Flask 应用程序进行很少的修改就可以变成 Quart 应用程序,然后就可以使用这些新的库来实现 Flask 无法做到的性能优化。

本文详细介绍了典型的生产环境的 CRUD 应用程序从 Flask 到 Quart 的转换,并展示相关的性能改进优势。

0. 我想直接看杰伦

将这个 Flask-pyscopg2 应用程序升级到 Quart-asyncpg 应用程序可以提高 3 倍的性能,而且不需要对代码进行重大的重写或调整。

如上所示,在针对单个资源详情的请求下,Flask每秒请求为330个,而quart能达到1160个。以此类推,Quart 相比于 Flask 平均性能提高3倍。

这个比较,我使用了一个简单的只提供一个 RESTful 接口的应用程序,这是微服务架构中的常见用例。

该应用程序有三个路由。这些路由分别是:

  • 单个电影详情:GET /films/pk/
  • 所有电影:GET /films/
  • 添加新评论:POST /reviews/

源代码可以在以下网址找到:
https://github.com/pgjones/faster_than_flask_article

有两个 commit ,分别是一个 Flask 版本和一个Quart 版本。

1. 从 Flask 到 Quart

从 Flask 改用 Quart 很容易,只需要一点点改变,特别是 from flask 改为 from quart,函数变成异步函数。

def add_review():
  data = request.get_json()
  ...

变成

async def add_review():
  data = await request.get_json()
  ...

2.数据库连接,从 psycopg2 到 asyncpg

从 psycopg2 改用 asyncpg 比较麻烦,因为两者有不同的用法。

为了简化区别,我们在 Flask 应用程序中使用了 PoolWrapper,使得 psycopg2 可以使用与 asyncpg 相同的 API 进行上下文管理,即:

with pool.acquire() as connection:

这将允许通过with更改为async with来使用asyncpg。

当然,除了连接之外,Asyncpg和psycopg2还在游标使用、事务、执行参数和查询格式方面存在差异。这些差异是你在迁移过程中需要注意的。

3.部署

Flask 应用程序往往不能直接在生产环境中直接暴露给用户,这是因为Flask 本身一次只能处理一个请求。因此,常常用WSGI服务器与某种异步 worker 结合使用,例如  带 eventlet 的Gunicorn。

Quart 也可以用 Gunicorn 部署,它允许使用相同的命令来运行 Flask 和 Quart 应用程序:

$ gunicorn --config gunicorn.py 'run:create_app()'
针对 Flask 和 Quart 的性能测试是基于 Gunicorn 进行的。

4.添加测试数据

除了添加一个简单的 review 表之外,Postgresql 示例数据库还要为应用程序提供一些用于 CRUD的数据。

CREATE TABLE review (
  film_id INTEGER REFERENCES film(film_id),
  rating INTEGER
);

5.性能测试

为了测量应用程序的性能,我们使用了wrk。它被配置为使用20个连接,以匹配数据库连接池的大小(确保最高的吞吐量,20是我使用过的典型值)。命令如下:

测试 GET 请求的命令是

$ wrk --connections 20 --duration 5m http://localhost:5000/${PATH}/

测试 POST 请求的命令是

$ wrk --connections 20 --duration 5m --script post.lua http://localhost:5000/${PATH}/

测试使用的 post.lua 文件如下:

wrk.method = "POST"
wrk.body = '{"film_id": 995, "rating": 4}'
wrk.headers["Content-Type"] = "application/json"

6.系统信息与结果

系统信息:

Postgres (9.5.10),wrk (4.0.0),Python (3.6.3),asyncpg (0.13.0),Flask (0.12.2),Gunicorn (19.7.1),psycopg2 (2.7.3.2), Quart (0.3.1)

全部运行在一台 AWS c4.large 机器上。

结果

请注意,Quart 服务器的平均等待时间减少了 2 至 3.5 倍,每秒的请求数量增加了 2 至 3.5 倍。

7.结论

Flask 应用程序升级到 Quart 应用程序是相当简单的,因为大部分 API 是共享的,所以主要工作就是在正确的位置写asyncawait。然而,如果使用 SQLAlchemy(或其他 ORM),则从psycopg2 到 asyncpg 的改变会比较复杂,并且可能会很麻烦。

这个 demo 应用程序的性能显着提高,这个改进主要是由于 Quart 使用了 asyncpg 和 uvloop,据估计,仅 Quart 就能提供 1.5 倍的提升。

总之,从 Flask-psycopg2 应用程序升级到 Quart-asyncpg 应用程序的比较简单,并拥有非常合理的性能改进。这可能会扩展到其他基于 asyncio 的库,意味着将 Flask 应用程序转换到 asyncio 生态系统,Quart 只需要很小的工作量。

原文:3x faster Flask apps
https://hackernoon.com/3x-faster-than-flask-8e89bfbe8e4f

我们的文章到此就结束啦,如果你喜欢今天的 Python 教程,请持续关注Python实用宝典。

有任何问题,可以在公众号后台回复:加群,回答相应验证信息,进入互助群询问。

原创不易,希望你能在下面点个赞和在看支持我继续创作,谢谢!

给作者打赏,选择打赏金额
¥1¥5¥10¥20¥50¥100¥200 自定义

​Python实用宝典 ( pythondict.com )
不只是一个宝典
欢迎关注公众号:Python实用宝典

人类训练AI的方式存在根本性缺陷

机器学习模型在实验室能被调整到近乎完美,但在现实环境中进行试验时,往往都会失败,这已经不是什么秘密了。这通常被归结为人工智能接受训练和测试的数据与它在世界上遇到的数据不匹配,这个问题被称为数据偏移(data shift)。

例如,经过训练,人类能够制造在高质量的医学图像中发现疾病迹象的AI,却难以分辨繁忙的诊所中廉价相机捕捉到的模糊或裁剪的图像。

这种现象被称为“数据不规范”(underspecification)。在机器学习世界中普遍存在。

数十名谷歌研究人员着眼于一系列不同的人工智能应用,从图像识别到自然语言处理(NLP)再到疾病预测。他们发现,“数据不规范”是这些项目表现不佳的原因。问题在于机器学习模型的训练和测试方式,没有简单的解决办法。

1.同样的模型,但表现不同

如果你阅读过我们前面的推文,你也知道建立一个机器学习模型需要在大量的数据源上进行训练,然后再在一堆模型没见过的数据源上进行测试,当模型通过测试时,模型就完成了。

谷歌人员指出,这样的判断标准太低了。事实上,训练过程可以产生许多不同的模型,即使这些模型都通过了测试,但他们内部存在许多细微的差异。比如神经网络不同节点的系数不同、被选择出来的关键因子不同。都会造成模型之间存在不同的差异。

这些差异,在模型通过测试的时候都会被忽略掉。而事实告诉我们,在现实世界中,这些细微的差异会导致巨大的表现差异。

这也就意味着,我们根本无法区分同一个算法训练出来的模型之间,哪个更适合在现实世界中运行,哪些模型根本不适合在现实生活中运行。

这与数据偏移不同,在数据偏移中,因为数据源和现实世界的实际情况不匹配,训练不能产生一个好的模型。但是在数据不规范中,你能产生许多通过测试的模型,但这些模型里掺杂着“好的模型”和“坏的模型”,它们只能通过在现实生活中实际应用来区分好坏。

研究人员研究了数据不规范对许多不同应用程序的影响,每种情况下,它们使用相同的算法训练产生多个机器学习模型,然后进行压力测试,突出它们在性能上的具体差异。

例如,ImageNet是一个日常物体的图像数据集,他们基于ImageNet上训练了50种版本的图像识别模型。每组训练的唯一区别是开始时分配给神经网络的随机值。然而,尽管所有50个模型在测试中的得分都差不多,但它们在压力测试中的表现却大相径庭。

压力测试使用ImageNet-C,这些图像被马赛克化或改变了亮度和对比度。另一个压力测试集是ObjectNet,一个日常物品的异常姿势图像集,比如倒立的茶壶,挂在钩子上的t恤。

50位模型里,有些在被马赛克化的图片上识别得很好,有些则在异常姿势上识别得很好;有些模型的整体表现比其他模型好得多。但其实就训练过程而言,它们都是一样的。

研究人员使用三种医学AI系统根据视网膜扫描预测眼病,从皮肤病变中预测癌症,从患者病历中预测肾衰竭。每个系统都有同样的问题:

那些本应同样精确的模型,在用真实世界的数据(如不同的皮肤类型)测试时,出现了不同的表现。这并不是训练集/测试集不够导致的,而是不论训练集/测试集多大都会出现这样的问题。

因此研究人员罗勒说,我们可能需要重新考虑如何评估神经网络。“我们的基本假设中出现了一些重大漏洞。”

人类现在使用的大多数机器学习模型的训练逻辑,其实都无法被证明是有效的。

2.解决方案

一种选择是在训练和测试过程中设计一个额外的阶段,在这个阶段中可以同时生产多个模型,而不是只生产一个。然后,这些相互竞争的模型可以在具体的现实任务中再次进行测试,以选择最适合这项工作的模型。

但这样需要做很多工作。苏黎世联邦理工学院的机器学习研究员Yannic Kilcher说,对于像谷歌这样建造和部署大型模型的公司来说,这样做是值得的。谷歌可以提供50种不同版本的NLP模型,应用程序开发人员可以选择最适合他们的一个。

目前研究人员们还没有解决这个问题,但正在探索改进培训过程的方法。当人工智能在现实世界中表现不佳时,人们就不太愿意使用它了。因此,如果这个问题没有尽早得到有效的解决,这个时代的人工智能浪潮或许将就此平息。

译自technologyreview, 《The way we train AI is fundamentally flawed》。

我们的文章到此就结束啦,如果你喜欢今天的Python 文章,请持续关注Python实用宝典。

有任何问题,可以在公众号后台回复:加群,回答相应验证信息,进入互助群询问。

原创不易,希望你能在下面点个赞和在看支持我继续创作,谢谢!


​Python实用宝典 (pythondict.com)
不只是一个宝典
欢迎关注公众号:Python实用宝典

这3种数组内的字典去重方式 你会几种?

你知道吗?如果数组是字典组成的,直接对数组内的字典采用set的方式进行去重,会报错:

test = [{"a": 1}, {"a": 1}, {"a": 3}, {"b": 4}]
test = list(set(test))
>>>TypeError: unhashable type: 'dict'

因为使用set去重的前提是该对象为不可变对象,而字典是可变对象,因此无法使用该方法去重。

那么怎么解决这个问题呢?有三个办法。

1.使用reduce方法

reduce() 函数会对参数序列中元素进行累积。

比如:

from functools import reduce
>>>def add(x, y) :            # 两数相加
...    return x + y
... 
>>>reduce(add, [1,2,3,4,5])   # 计算列表和:1+2+3+4+5
15

上述写法也能用lambda函数简化为:

from functools import reduce
>>> reduce(lambda x, y: x+y, [1,2,3,4,5])  # 使用 lambda 匿名函数
15

因此,我们自己编写一个函数进行数组内的字典去重:

from functools import reduce

data = [{"a": 1}, {"a": 1}, {"a": 3}, {"b": 4}]
result = []
def unduplicate(result, data):
    if data not in result:
        result = result + [data]
    return result

for i in data:
    result = unduplicate(result, i)

>>> result
>>> [{'a': 1}, {'a': 3}, {'b': 4}]

稍显复杂,如果使用reduce函数和lambda函数,代码能简化很多:

def delete_duplicate(data):
    func = lambda x, y: x + [y] if y not in x else x
    data = reduce(func, [[], ] + data)
    return data

>>> delete_duplicate(data)
>>> [{'a': 1}, {'a': 3}, {'b': 4}]

当然, 我也能一行写完这个功能:

data = reduce(lambda x, y: x + [y] if y not in x else x, [[], ] + data)

只不过有可能会被打死在工位上,所以不建议这么干。

2.奇怪的技巧

就如文章开头提到的,字典之所以不能用set去重,是因为它是可变对象。

但是…如果我们把它变成不可变对象呢?

data = [{"a": 1}, {"a": 1}, {"a": 3}, {"b": 4}]
def delete_duplicate(data):
    immutable_dict = set([str(item) for item in data])
    data = [eval(i) for i in immutable_dict]
    return data
>>> delete_duplicate(data)
>>> [{'a': 1}, {'a': 3}, {'b': 4}]

没错,这能成。

1.遍历字典,将每个子项变成字符串存放到数组中,再通过set函数去重。

2.通过eval函数,将去重后的数组里的每个子项重新转化回字典。

如此Python,怎能不好玩?

3.高效的方式

上面讲了两种骚操作,其实都不太建议在实际工作中使用。

一个原因是真的太骚了,怕被打趴在工位上。

另一个原因是,它们在应对较大数据量的时候,性能不太行。

下面是最正统的方式:

data = [dict(t) for t in set([tuple(d.items()) for d in data])]
>>>data
>>>[{'a': 1}, {'b': 2}]

其实和第二种方式一样,是将数组内的每个字典转成元组,也就是不可变对象,再使用set进行去重。去重完毕后再使用dict函数将元组重新组成字典对。

但是,这种方法对于字典内还有字典的数据结构是不适用的,因此对于字典对里还有字典情况的去重,比如:

data2 = [{"a": {"b": "c"}}, {"a": {"b": "c"}}]

这种情况我建议使用第二种方式去重:

data2 = [{"a": {"b": "c"}}, {"a": {"b": "c"}}]
def delete_duplicate_str(data):
    immutable_dict = set([str(item) for item in data])
    data = [eval(i) for i in immutable_dict]
    return data
print(delete_duplicate_str(data2))

>>> [{'a': {'b': 'c'}}]

怎么样,这三种方式你都学会了吗?如果觉得有收获的话记得收藏一下。以后遇到类似的去重场景时可以拿出阅读一下。

我们的文章到此就结束啦,如果你喜欢今天的 Python 教程,请持续关注Python实用宝典。

有任何问题,可以在公众号后台回复:加群,回答相应验证信息,进入互助群询问。

原创不易,希望你能在下面点个赞和在看支持我继续创作,谢谢!

给作者打赏,选择打赏金额
¥1¥5¥10¥20¥50¥100¥200 自定义

​Python实用宝典 ( pythondict.com )
不只是一个宝典
欢迎关注公众号:Python实用宝典

为什么小黄鸭调试法在中国行不通

许多程序员都有向他人请教代码问题或解释代码的经历,而在解释的过程中,程序员可能就发觉了问题的解决方案,一边阐述代码的意图,一遍观察代码实际上产生的行为并调试,一旦有任何不协调的地方就能够迅速地发现并解决问题。

小黄鸭调试法就是这样,工作的时候,在电脑旁边放一只小黄鸭,耐心地跟它讲解每一行代码的逻辑和意义,以此来激发灵感和发现矛盾。

可惜,在当今中国的编码环境中,由于产品迭代周期极短,需求量大、代码量大,程序员根本没有时间在产品上线前实施小黄鸭调试。

而这样就引发了一个问题,代码没有经过小黄鸭调试,可能存在许多隐藏的Bug,而在程序员修复这些隐藏的Bugs的时候,可能会引入新的Bug…

更糟糕的是,如果修Bug的人不是当初写Bug的那个人,对整个代码逻辑和功能逻辑并不一定非常了解,那么有可能写出比原有Bug更加严重的Bug…

久而久之就形成了负反馈,产品功能越来越多,程序员越来越多,代码越写越糟糕,最终达到一种状态:能用就行

那么为什么?小黄鸭调试法在中国行不通的本质原因是什么?

1.“时间就是金钱,效率就是生命”

房价太高,结婚要买房,娃儿要读书,不急不行。上头急、组长急、产品急、开发急,大家都急,急急如厕令,厕所都得装个计时器,哪有时间跟你只鸭子折腾。

2.代码写得好,加班少,涨薪真的没你份儿

大部分开发者的目的都不是写出最优质的代码,而是赚最多的钱。为了钱而写代码,编程不过赚钱的工具,功能写得越多、钱就越多。

Bug多一点无所谓,毕竟老子写的功能比Bug多得多,那你小黄鸭调试法还有啥用,不是在拖累我赚钱的速度吗?

3.各种编程营销组织带坏编程风气

如果你是编程教育的细分领域,教量化投资、SaaS编程等技术含量高的课程也就算了。问题是一个教Python基础的课程起这样的标题是何居心?

总而言之,小黄鸭调试法的行不通,是因为中国程序员大量的时间(包括加班时间),花在了一些原本可以避免的Bug和可有可无的需求上。

从微观上看,是对KPI的焦虑、是对生活的担忧。

从宏观上看,是公司违法成本低、甚至根本不需要成本。

租下来的办公楼,多开几个小时灯就能多出一个新功能,成本近乎于零,没人用不亏,有人用则赚爆,何乐而不为?

但是许多人没想过的是,这样的生产成果,终究会化为垃圾,甚至对于整个产品而言单纯只是拖油瓶。真正长久不衰的功能,往往需要精心打造,而非急功近利。

在当今中国的编码环境中,我看不到有进行任何科学规划的团队,看不到有精心打造软件产品的公司。

就长期而言,我对中国的软件业持看空态度。为了急功近利的眼前效益,程序员生产的代码质量自然变得极差,就如同前面提到的负反馈效应,越往后,质量只会越来越差。

代码世界里的那些令人喷饭的注释

代码里的注释经常能让人嘀笑皆非,有些人喜欢在里边“搞事情”,另外有些人非常擅长写幽默搞笑的注释内容,还有些人无奈地在注释里告警后人…

下面就给大家展示一些国外程序员们写的注释。

注意:看的时候严禁喝水或进食。

1、亲爱的代码维护人员:

当您尝试优化这段代码但发现这是一个极端错误的决定的时候,请修改下面的计时器,以便警示后人。

总计浪费在这段代码的时间 = 16 小时

2、真的很有问题

3、谨以此代码献给我的妻子达琳,感谢她一直支持我,还有我三个孩子和一只狗。

4、神奇代码,请勿改动

5、喝醉啦,迟些再弄

6、你可能会认为你读得懂以下的代码。但是你不会懂的,相信我吧。

要是你尝试玩弄这段代码的话,你将会在无尽的通宵中不断地咒骂自己为什么会认为自己聪明到可以优化这段代码。

好了,现在请关闭这个文件去玩点别的吧。

7、程序员1(于 2002 年 6 月 7 日):在登陆界面临时加入一些调试代码

程序员2(于 2007 年 5 月 22 日):临你个屁啊

8、反正这个办法就修复了问题,我也不知道为什么会这样

9、要理解什么是递归的话,请参考本文件的底部

(在文件的底部)

要理解什么是递归的话,请参考本文件的顶部

10、狂插两下; //痛啊

11、亲爱的未来的我自己,请原谅我。

我有着难以表达的歉意。

12、我不对以下代码负责。

是他们逼我写的,是违背我意愿的。

13、疯了吗?欢迎来到斯巴达。

14、要是你能修正这个问题的话,我会送给你两个七十二岁的处女

15、没有注释留给你,难写的代码必定难读

16、IE 浏览器的 Hack (在这里先假设 IE 是浏览器)

17、有待修正。 修正什么啊?

18、要是再让我看到这种代码,我会带着枪来上班的

19、有只龙在这里……

20、在你阅读以下代码时,你要先搞懂为什么我在这样做。

我想读取一个根节点下面所有的子节点,以便控制根节点不会显示在选择框上。但那个傻逼的 DBA 找了一些某些傻逼的借口不让我用索引去读取这些数据,而要求我用他们傻逼的迭代器。所以有了以下代码。

21、当我写这段代码的时候,只有老天和我自己知道我在做什么。

现在,只剩老天知道了。

我们的文章到此就结束啦,如果你喜欢今天的 Python 教程,请持续关注Python实用宝典。

有任何问题,可以在公众号后台回复:加群,回答相应验证信息,进入互助群询问。

原创不易,希望你能在下面点个赞和在看支持我继续创作,谢谢!

给作者打赏,选择打赏金额
¥1¥5¥10¥20¥50¥100¥200 自定义

​Python实用宝典 ( pythondict.com )
不只是一个宝典
欢迎关注公众号:Python实用宝典

自定义MySQL数据流 — Python 量化投资实战教程(9)

前面八篇量化投资实战教程,我们所使用到的数据仅仅只有收盘价、成交量等普通指标,如果我们有其他的指标需要进行回测怎么办?

此外,前面使用的数据源都是基于csv文件的,我们能否从数据库(比如MySQL)中直接提取数据作为回测的数据源呢?

​事实上,backtrader虽然没有直接提供接口给我们做这样的优化,但是我们可以通过继承DataBase基类重写DataFeed实现目的。下面就给大家演示一下如何从MySQL中提取数据并增加换手率指标进行回测。

本文完整源代码和数据均在开源代码仓库中:
https://github.com/Ckend/pythondict-quant

1.准备

开始之前,你要确保Python和pip已经成功安装在电脑上,如果没有,请访问这篇文章:超详细Python安装指南 进行安装。

(可选1) 如果你用Python的目的是数据分析,可以直接安装Anaconda:Python数据分析与挖掘好帮手—Anaconda,它内置了Python和pip.

(可选2) 此外,推荐大家用VSCode编辑器来编写小型Python项目:Python 编程的最好搭档—VSCode 详细指南

Windows环境下打开Cmd(开始—运行—CMD),苹果系统环境下请打开Terminal(command+空格输入Terminal),输入命令安装依赖:

在终端输入以下命令安装我们所需要的依赖模块:

pip install backtrader
pip install numpy
pip install matplotlib

看到 Successfully installed xxx 则说明安装成功。

2.自定义DataFeed

何为DataFeed?DataFeed 即 backtrader 中的“数据源”,任何数据进入策略回测前都要通过DataFeed,而DataFeed中会对数据进行处理,使得策略可以高效地进行计算。

而我们今天要做的,就是增加一个基于MySQL的DataFeed,使得整个流程变得更加自动化。

首先,需要定义一个类,使其继承与backtrader的数据基类 DataBase:

from backtrader.feed import DataBase
from backtrader import date2num


class MySQLData(DataBase):
    pass

如果需要从外部传入所需股票数据的代码和其一定范围内的K线数据,需要提前定义params. 此外,如果你有除了datetime(时间)open(开盘价)close(收盘价)high(最高价)low(最低价)volume(成交量) 之外的指标。需要提前定义lines,如下所示:

class MySQLData(DataBase):
    params = (
        ('fromdate', datetime.datetime.min),
        ('todate', datetime.datetime.max),
        ('ts_code', ''),
    )
    lines = (
        "turnover",
        "turnover_rate"
    )

可见在lines中,我增加了两个指标:turnover(成交额)turnover_rate(换手率)

接下来,编写一个函数根据params参数从MySQL中获取数据:

    def load_data_from_db(self, table, ts_code, start_time, end_time):
        """
        从MySQL加载指定数据
        Args:
            table (str): 表名
            ts_code (str): 股票代码
            start_time (str): 起始时间
            end_time (str): 终止时间
        return:
            data (List): 数据集
        """
        db = pymysql.connect(
            host="localhost",
            user="root",
            password="12345678",
            db="golden_stone",
            port=3306
        )

        cur = db.cursor()
        sql = cur.execute(
            f"SELECT * FROM {table} WHERE trade_time >= '{start_time}' and trade_time < '{end_time}'"
            f"and ts_code = '{ts_code}' order by trade_time asc"
        )
        data = cur.fetchall()
        db.close()
        return iter(list(data))

代码本身没有什么可说的,记得替换你本地的mysql配置,值得注意的是最后一行,拿到mysql数据后需要转化为迭代器。

在类初始化的时候,需要定义相关的数据存放变量并调用上述函数获取数据:

    def __init__(self, *args, **kwargs):
        self.result = []
        self.empty = False

    def start(self):
        self.result = self.load_data_from_db("stock_normalk", self.p.ts_code, self.p.fromdate, self.p.todate)

接下来到了关键的步骤,在 `cerebro.adddata(data)` 的时候,cerebro会遍历Datafeed的所有数据,此时会调用_load函数, 因此我们需要在这里,将数据库中提取的每列数据对应到lines上:

    def _load(self):
        if self.empty:
            return False
        try:
            one_row = next(self.result)
        except StopIteration:
            return False
        self.lines.datetime[0] = date2num(one_row[3])
        self.lines.open[0] = float(one_row[4])
        self.lines.high[0] = float(one_row[5])
        self.lines.low[0] = float(one_row[6])
        self.lines.close[0] = float(one_row[7])
        self.lines.volume[0] = float(one_row[8])
        self.lines.turnover[0] = float(one_row[9])
        self.lines.turnover_rate[0] = float(one_row[12])
        return True

如果你完整地看完了我的上述分析,那么理解下面整个DataFeed,甚至自己写一个DataFeed,是非常容易的。

# Python 实用宝典
# 自定义数据流 — Python 量化投资实战教程(9)
import datetime
import traceback
import pymysql

from backtrader.feed import DataBase
from backtrader import date2num


class MySQLData(DataBase):
    params = (
        ('fromdate', datetime.datetime.min),
        ('todate', datetime.datetime.max),
        ('ts_code', ''),
    )
    lines = (
        "turnover",
        "turnover_rate"
    )

    def load_data_from_db(self, table, ts_code, start_time, end_time):
        """
        从MySQL加载指定数据
        Args:
            table (str): 表名
            ts_code (str): 股票代码
            start_time (str): 起始时间
            end_time (str): 终止时间
        return:
            data (List): 数据集
        """
        db = pymysql.connect(
            host="localhost",
            user="root",
            password="12345678",
            db="golden_stone",
            port=3306
        )

        cur = db.cursor()
        sql = cur.execute(
            f"SELECT * FROM {table} WHERE trade_time >= '{start_time}' and trade_time < '{end_time}'"
            f"and ts_code = '{ts_code}' order by trade_time asc"
        )
        data = cur.fetchall()
        db.close()
        return iter(list(data))

    def __init__(self, *args, **kwargs):
        self.result = []
        self.empty = False

    def start(self):
        self.result = self.load_data_from_db("stock_normalk", self.p.ts_code, self.p.fromdate, self.p.todate)

    def _load(self):
        if self.empty:
            return False
        try:
            one_row = next(self.result)
        except StopIteration:
            return False
        self.lines.datetime[0] = date2num(one_row[3])
        self.lines.open[0] = float(one_row[4])
        self.lines.high[0] = float(one_row[5])
        self.lines.low[0] = float(one_row[6])
        self.lines.close[0] = float(one_row[7])
        self.lines.volume[0] = float(one_row[8])
        self.lines.turnover[0] = float(one_row[9])
        self.lines.turnover_rate[0] = float(one_row[12])
        return True

3.使用自定义数据流进行回测

接下来,让我们尝试使用这个自定义数据流输入数据,采用第二章的macd策略辅助增加换手率指标进行回测。

这里当然需要你先读懂第二章的内容,如果有点忘记了,可以回头阅读一下,非常简单:

Python 量化投资实战教程(2) —MACD策略

首先,在回测模块及next函数中,引入换手率指标:

class TestStrategy(bt.Strategy):
    def log(self, txt, dt=None):
        ''' Logging function fot this strategy'''
        dt = dt or self.datas[0].datetime.date(0)
        print('%s, %s' % (dt.isoformat(), txt))

    @staticmethod
    def percent(today, yesterday):
        return float(today - yesterday) / today

    def __init__(self):
        self.dataclose = self.datas[0].close
        self.volume = self.datas[0].volume
        # 新的变更:引入换手率指标
        self.turnover_rate = self.datas[0].turnover_rate

next函数买入时增加判断换手率必须小于3%的条件:

    # Python 实用宝典
    def next(self):
        self.log('Close, %.2f' % self.dataclose[0])
        if self.order:
            return

        if not self.position:
            condition1 = self.macd[-1] - self.signal[-1]
            condition2 = self.macd[0] - self.signal[0]
            # 增加判断换手率小于3%的条件
            if condition1 < 0 and condition2 > 0 and self.turnover_rate[0] < 3:
                self.log('BUY CREATE, %.2f' % self.dataclose[0])
                self.order = self.buy()

        else:
            condition = (self.dataclose[0] - self.bar_executed_close) / self.dataclose[0]
            if condition > 0.1 or condition < -0.1:
                self.log('SELL CREATE, %.2f' % self.dataclose[0])
                self.order = self.sell()

最后,引入我们刚刚编写完成的MySQLData Feed,传入相关参数读取股票为603520.SH的数据流,取2017年1月1日至2020年4月12日的数据,并调用回测函数:

if __name__ == '__main__':
    cerebro = bt.Cerebro()

    cerebro.addstrategy(TestStrategy)

    # 加载数据到模型中
    data = MySQLData(
        ts_code="sh603520",
        fromdate=datetime.datetime(2017, 1, 1),
        todate=datetime.datetime(2020, 4, 12),
    )
    cerebro.adddata(data)

    cerebro.broker.setcash(10000)

    cerebro.addsizer(bt.sizers.FixedSize, stake=100)

    cerebro.broker.setcommission(commission=0.005)

    print('Starting Portfolio Value: %.2f' % cerebro.broker.getvalue())

    cerebro.run()

    print('Final Portfolio Value: %.2f' % cerebro.broker.getvalue())

    cerebro.plot()

效果如下:

与原来相比,加入换手率指标后,利润率有一定的提高,不过就如我们之前所说的,单一股票维度的回测是不准确的,如果大家有兴趣,可以将第三章:

Python 量化投资实战教程(3) —A股回测MACD策略

改造一下并加入换手率指标进行回测,看看这个指标是否真的有正效益。

本文完整源代码和数据均在开源代码仓库中:
https://github.com/Ckend/pythondict-quant

如果你访问不了github,也可以在公众号后台回复 量化投资9 下载相关代码。

欢迎在公众号后台回复:加群,回答相应红字验证信息,进入互助群交流。

我们的文章到此就结束啦,如果你喜欢今天的Python 实战教程,请持续关注Python实用宝典。

原创不易,希望你能在下面点个赞和在看支持我继续创作,谢谢!


​Python实用宝典 (pythondict.com)
不只是一个宝典
欢迎关注公众号:Python实用宝典

Python编写的超帅数独可视化解题器

数独相信大家都玩过,它被称为“聪明人的游戏”,在很多人眼里:

会玩数独=高智商

为什么?因为数独能够培养观察力,提高反应力: 数独的练习能够锻炼手眼脑的协调性、提高手脑并用的能力,锻炼大脑的思维灵活度,全面提高反应力。

非常适合孩子在成长过程中锻炼大脑,适合成年人在生活中激活思维。

不过当我们遇到不会解的数独怎么办?答案是,用Python算出来!

基于 Pygame-Sudoku-Solver 这个开源项目,可视化解决数独问题变得极其简单。

1.准备

开始之前,你要确保Python和pip已经成功安装在电脑上,如果没有,请访问这篇文章:超详细Python安装指南 进行安装。

(可选1) 如果你用Python的目的是数据分析,可以直接安装Anaconda:Python数据分析与挖掘好帮手—Anaconda,它内置了Python和pip.

(可选2) 此外,推荐大家用VSCode编辑器来编写小型Python项目:Python 编程的最好搭档—VSCode 详细指南

Windows环境下打开Cmd(开始—运行—CMD),苹果系统环境下请打开Terminal(command+空格输入Terminal),输入命令安装依赖:

1.在终端输入以下命令下载该开源库

git clone https://github.com/tymscar/Pygame-Sudoku-Solver.git

2.使用cd命令进入该文件夹,并安装依赖:

cd Pygame-Sudoku-Solver
pip install -r requirements.txt

接下来,可以试试运行该项目了:

python solver.py

此时会出现一个空白3*3的九宫格

2.怎么解题

这个开源项目的解题方法如下:

1.输入题目数字 — 你只需要点击空白区域,此时会回显绿色方块,输入数字,如果数字合法则会填入框内,如果不合法则会闪现红色。

2.当你将数独题目里的所有数字填写完毕,单击空格键即可开始运算:

而且,细心的作者还帮大家准备了夜晚模式,单击“d”键可切换到夜晚模式:

3.原理

所有的解题源代码都放在了solver.py文件中,大家可以在里面看到整个解题过程。

作者没有写任何注释,但是代码逻辑思路是清晰的,比如核心判断逻辑,Cell类里的 isValid, 用于判断某个值 (what变量) 放进某个 Cell 里是否合法:

此处,lineV.cells 表示数组中每一列组成的cell;lineH.cells即每一行组成的cell;box.cells即每个子九宫格。他们都有一个共同的特点:其中不能出现重复的值。

因此你会看到如果某个值存在于这些cells当中,isValid直接返回False,表明其不应该出现在这个位置。

如果你的网络较差,git clone拿不到代码,可以在公众号后台回复:数独 下载源代码。

我们的文章到此就结束啦,如果你喜欢今天的 Python 教程,请持续关注Python实用宝典。

有任何问题,可以在公众号后台回复:加群,回答相应验证信息,进入互助群询问。

原创不易,希望你能在下面点个赞和在看支持我继续创作,谢谢!

给作者打赏,选择打赏金额
¥1¥5¥10¥20¥50¥100¥200 自定义

​Python实用宝典 ( pythondict.com )
不只是一个宝典
欢迎关注公众号:Python实用宝典

Python 自动化,Helium 凭什么取代 Selenium?

来自AirPython哥的分享。

1. Helium 是什么?

Helium 是一款 Web 端自动化开源框架,全称是:Selenium-Python-Helium,从名字上就可以看出,Helium 似乎和 Selenium 息息相关

确实,Helium 针对 Selenium 进行了封装,它屏蔽了 Selenium 很多实现细节,提供了更加简洁直观的 API,更方便我们进行 Web 端的自动化

官方表示,要实现同样的功能,Helium 相比 Selenium 要少 30% – 50% 的代码

目前,Helium 仅支持 Chrome 和 FireFox

2.优缺点

Helium 主要包含下面 6 个优点:

  • Helium 自带 WebDriver,不需要下载、配置浏览器驱动
  • 内嵌页面 iframe 页面元素直接操作,不需要使用 switch_to.frame() 切换 iframe
  • 窗体管理更方便,可以直接使用窗口标题或部分标题内容来切换窗体
  • 隐式等待,针对某个元素执行点击操作,Selenium 如果元素没有出现,脚本会执行失败;而 Helium 默认最多等待 10s,等待元素出现后立马执行点击操作
  • 显式等待,Helium 提供更加优雅的 API 来等待页面元素出现
  • API 更简洁直观,代码量少

Helium 主要缺点,体现在:

  • 由于封装,屏蔽了很多细节,所以它不合适二次开发
  • 目前仅支持 Chrome 和 FireFox 浏览器
  • 版本更新慢、遗留 Bug 及文档少

3.准备一下

切换到对应的虚拟环境下,通过 pip 命令安装依赖即可

# 安装依赖
pip3 install helium

接着,我们在 IDE 中,使用 helium.__all__ 打印出它包含的属性及方法

我们发现,Helium 包含的操作动作、控件对象、键盘操作关键字基本覆盖了大部分的自动化操作场景

4.Selenium VS Helium

是骡子是马,拉出来溜溜 ~

接下来,我们以登录 126 邮箱为例,来比较 Selenium 和 Helium

1、传统 Selenium 实现

首先,我们需要下载并配置 WebDriver,然后实例化 WebDriver 对象,打开邮箱登录的主页面

from selenium import webdriver

# 实例化Driver
driver = webdriver.Chrome()
# 隐式等待10s
driver.implicitly_wait(10)
# 打开主页面
driver.get(home_url)

通过观察网页元素,发现输入框区域被包裹在 iframe 内嵌页面中

所以,我们需要使用 switch_to.frame() 函数切换到对应的 iframe,才能操作 iframe 内部的元素

from selenium.webdriver.common.by import By
from selenium.webdriver.support import expected_conditions as EC
from selenium.webdriver.support.ui import WebDriverWait

# 显示等待打开主页面
wait = WebDriverWait(driver, 10, 0.5)

# 切换到对应的iframe,否则无法操作内部元素
wait.until(
    EC.frame_to_be_available_and_switch_to_it(driver.find_element_by_xpath('//iframe[contains(@id,"x-URS-iframe")]')))

接着,使用 Selenium API( 这里以 Xpath 为例 )查找用户名和密码输入框、登录按钮;输入用户名和密码,模拟登录操作

# 用户名输入框
element_input = wait.until(EC.visibility_of(driver.find_element_by_xpath('//input[@name="email"]')))
element_input.clear()
element_input.send_keys(username)

# 密码输入框
element_password = wait.until(EC.visibility_of(driver.find_element_by_xpath('//input[@name="password"]')))
element_password.clear()
element_password.send_keys(password)

# 登录按钮
wait.until(EC.element_to_be_clickable((By.XPATH, '//a[@id="dologin"]'))).click()

最后,通过某个页面元素是否出现来判断是否登录成功

# 找一个登录成功的页面元素
# 通过元素属性+元素值来唯一定位元素
result = True
try:
    element_recy_email = wait.until(EC.element_to_be_clickable((By.XPATH, '//span[@class="oz0" and contains(text(),"收 信")]')))
    if element_recy_email:
        result = True
    else:
        result = False
except Exception as e:
    result = False

print("邮箱登陆成功" if result else "邮箱登录失败")

2、Helium 实现

接下来,我们通过 Helium 的方式来实现这一操作

首先,我们只需要 2 行代码即可以打开主页

from helium import *

# 打开主页
driver = start_chrome(home_url)

# 等待元素加载完成
wait_until(Text("你的专业电子邮局").exists)

然后,通过内置 TextField 控件对象及预设文本内容,使用 write 动作输入用户名和密码

# 不需要切换iframe,直接输入
write(username,TextField('邮箱帐号或手机号码'))
write(password,TextField('输入密码'))

值得一提的是,Helium 不需要切换 iframe,可以直接操作内嵌页面元素,简直不要太方便!

# 模拟点击Enter键登录
press(ENTER)

通过 Helium 内置的 wait_until 方法 + 控件对象,可以显式等待元素出现,默认最长时间为 10s比如,这里等待登录完成的主页面加载完成,收件箱可以点击,执行一次点击操作

wait_until(Text('收 信').exists)

# 点击收件箱
click(Text('收 信'))

最后,调用 kill_browser() 方法关闭浏览器,结束自动化操作

# 退出
sleep(10)

# 关闭浏览器
kill_browser()

需要指出的是,Helium 使用 start_chrome() 方法返回的对象实际上就是 WebDriver 对象,可以结合 Selenium API 一起使用

5.最后

通过上面的对比发现,Helium 相比 Selenium 使用似乎更方便,但是它不适用于一些复杂的页面

因此,在实际自动化项目中,建议搭配 Selenium 和 Helium 使用,简单的页面使用 Helium,复杂的页面切换到 Selenium

如果你觉得文章还不错,请大家 点赞、分享、留言下,因为这将是我持续输出更多优质文章的最强动力!

Python制作国际空间站实时跟踪器

Open Notify是一个开源项目,旨在为NASA的一些出色数据提供简单的编程接口。

open-notify.org 的作者做了一些工作,以获取原始数据并将其转换为与太空和航天器有关的API

本文将通过这个接口,获取得到国际空间站的位置,并实时地绘制到地图上:

感谢cr0sis/Real-time-International-space-station-tracker

为了实现本文的目标,你得先安装ISS_Info:

pip install ISS-Info

下面分步骤讲解整套绘制流程

1.地图初始化

为了实时展示国际空间站的路径,需要使用turtle绘制曲线,因此可以创建一个turtle画布,将背景设为地球:

  
import ISS_Info
import turtle
import time
import json
import urllib.request

screen = turtle.Screen()
screen.setup(720,360)
screen.setworldcoordinates(-180,-90,180,90)
screen.bgpic("map.png")
screen.bgcolor("black")
screen.register_shape("isss.gif")
screen.title("Real time ISS tracker")

iss = turtle.Turtle()
iss.shape("isss.gif")

2.获取空间站的人数

如果能知道空间站上的宇航员人数,我们就能更加准确的跟踪国际空间站。幸运的是open-notify确实提供了这样的接口。

为了获取人数信息,我们必须向:
http://api.open-notify.org/astros.json
请求拿到数据,并将相应的宇航员名字写在左上角:

astronauts = turtle.Turtle()
astronauts.penup()
astronauts.color('black')
astronauts.goto(-178,86)
astronauts.hideturtle()
url = "http://api.open-notify.org/astros.json"
response = urllib.request.urlopen(url)
result = json.loads(response.read())
print("There are currently " + str(result["number"]) + " astronauts in space:")
print("")
astronauts.write("People in space: " + str(result["number"]), font=style)
astronauts.sety(astronauts.ycor() - 5)

people = result["people"]

for p in people:
    print(p["name"] + " on: " + p["craft"])
    astronauts.write(p["name"] + " on: " + p["craft"], font=style)
    astronauts.sety(astronauts.ycor() - 5)

3.绘制空间站位置

为了能够绘制空间站的实时位置,我们需要请求拿到空间站的位置信息。请求的接口是:
http://api.open-notify.org/iss-now.json

不过作者将其封装成了一个函数,我们直接调用 iss_current_loc 即可,循环获取国际空间站位置:

while True:  
    location = ISS_Info.iss_current_loc()
    lat = location['iss_position']['latitude']
    lon = location['iss_position']['longitude']
    print("Position: \n latitude: {}, longitude: {}".format(lat,lon))
    pos = iss.pos() 
    posx = iss.xcor()
    if iss.xcor() >= (179.1):           ### Stop drawing at the right edge of  
        iss.penup()                     ### the screen to avoid a 
        iss.goto(float(lon),float(lat)) ### horizontal wrap round line
        time.sleep(5)
    else:
      iss.goto(float(lon),float(lat))
      iss.pendown()
      time.sleep(5)

我们还可以标出自己目前所处的位置,以查看和国际空间站的距离及空间站经过你上空的时间点(UTC)。

# 深圳
lat = 112.5118928
lon = 23.8534489

prediction = turtle.Turtle()
prediction.penup()
prediction.color('yellow')
prediction.goto(lat, lon)
prediction.dot(5)
prediction.hideturtle()

url = 'http://api.open-notify.org/iss-pass.json?lat=' +str(lat-90) + '&lon=' + str(lon)
response = urllib.request.urlopen(url)
result = json.loads(response.read())

over = result ['response'][1]['risetime']

prediction.write(time.ctime(over), font=style) 

不过这里值得注意的是,iss-pass.json这个接口的纬度计算必须在-90到90之内,因此深圳的纬度需要减去90.

最终效果如下:

在Python实用宝典公众号后台回复“国际空间站”或者“ISS”即可获得本文完整源代码哦。

我们的文章到此就结束啦,如果你喜欢今天的 Python 教程,请持续关注Python实用宝典。

有任何问题,可以在公众号后台回复:加群,回答相应验证信息,进入互助群询问。

原创不易,希望你能在下面点个赞和在看支持我继续创作,谢谢!

给作者打赏,选择打赏金额
¥1¥5¥10¥20¥50¥100¥200 自定义

​Python实用宝典 ( pythondict.com )
不只是一个宝典
欢迎关注公众号:Python实用宝典

Python 导出word所有图片并转化格式

作者:叶庭云

日常工作中,你是否遇到过这样的场景,领导发来一份 Word 文档,要求你将文档中的图片存储到一个文件夹内,并且还要将图片都改成 .jpg 或者 .png,你会怎么办?

你是不是一边内心崩溃,一边开始一张张的 另存为

今天,我就教你两招省时省力的方法,不管文档中有几张甚到几百张图片,你都可以快速保存下来。

一、分析

图片在文档的应用已经是十分普遍的现象了,在Word文档中插入合适的图片无疑会让我们的文档变得更美观。

先来回想一下,我们平常是如何在Word中插入图片的?

  • 在本地电脑中事先存储好需要的图片素材,然后插入到Word中

其实,第二种方法有一个弊端在于图片只存在 Word 中,如果我们需要将它们保存到本地电脑中以供日后使用,最常用的方法是单击鼠标右键,选择 另存为图片,然后选择路径进行保存。

这种方法在只需要处理少数几张图片时还算适用,一旦图片数量增多,处理工作就会变得繁琐且容易出错。

那么,我们怎样可以将这些图片批量保存呢?

二、提取出 Word 文档里的图片

解决方法就是:更改文件格式。直接将 Word 文档的后缀名改成 “.rar” (“.zip”也是可以的)的压缩格式。打开压缩文件,点击【word】-【media】,文档中使用的图片就出现在这里,只需要选中解压出来即可。

用于测试的 Word 文档如下:

操作方法如下:

点击查看,选择详细信息,勾上文件扩展名。

直接将 Word 文档的后缀名改成 “.rar” (“.zip”也是可以的)的压缩格式。

打开压缩文件,点击【word】-【media】,文档中使用的图片就出现在这里,只需要选中解压出来即可。

三、利用 python 批量转换格式

# -*- coding: UTF-8 -*-
"""
@File    :test_01.py
@Author  :叶庭云
@CSDN    :https://yetingyun.blog.csdn.net/
"""
# 导入os模块
import os

# 不存在 jpg图片 这个文件夹  创建
if not os.path.exists('jpg图片'):
    os.mkdir('jpg图片')


path = r'.\jpg图片'
# 列出 media 文件夹下所有图片
files = os.listdir(r'.\media')

for item in files:
    # 拼接出media 文件夹下所有图片路径
    file_1 = '.\media' + '/' + item
    # 读取图片数据
    with open(file_1, 'rb') as f:
        con = f.read()
    # 重新写入  以 .jpg 格式 并保存到jog图片文件夹
    file_name = path + '/' + item.split('.')[0] + '.jpg'
    with open(file_name, 'wb') as f:
        f.write(con)

运行效果如下:

程序运行,嗖的一下,图片格式都转换成了 .jpg 并保存到新的文件夹里。

作者:叶庭云

CSDN:https://blog.csdn.net/fyfugoyfa

我们的文章到此就结束啦,如果你喜欢今天的 Python 教程,请持续关注Python实用宝典。

有任何问题,可以在公众号后台回复:加群,回答相应验证信息,进入互助群询问。

原创不易,希望你能在下面点个赞和在看支持我继续创作,谢谢!

给作者打赏,选择打赏金额
¥1¥5¥10¥20¥50¥100¥200 自定义

​Python实用宝典 ( pythondict.com )
不只是一个宝典
欢迎关注公众号:Python实用宝典

有趣好用的Python教程

退出移动版
微信支付
请使用 微信 扫码支付