python

python

B站批量下载某个UP主的所有视频

python李魔佛 发表了文章 • 0 个评论 • 247 次浏览 • 2022-05-21 18:48 • 来自相关话题

B站上不少优秀的学习资源,下载到本地观看,便于快进,多倍速。 也可以放到平板,手机,在没有网络,或者网络条件不佳的环境下观看。
 

 
使用python实现
https://github.com/Rockyzsu/bilibili
 
B站视频下载
自动批量下载B站一个系列的视频

下载某个UP主的所有视频

使用:

下载you-get库,git clone https://github.com/soimort/you-get.git 复制其本地路径,比如/root/you-get/you-get

初次运行,删除history.db 文件, 修改配置文件config.py

START=1 # 下载系列视频的 第一个
END=1 # 下载系列视频的最后一个 , 比如一个系列教程有30个视频, start=5 ,end = 20 下载从第5个到第20个
ID='BV1oK411L7au' # 视频的ID
YOU_GET_PATH='/home/xda/othergit/you-get/you-get' # 你的you-get路径
MINS=1 # 每次循环等待1分钟
user_id = '518973111' # UP主的ID
total_page = 3 # up主的视频的页数
执行 python downloader.py ,进行下载循环

python people.py ,把某个up主的视频链接加入到待下载队列

python add_data.py --id=BV1oK411L7au --start=4 --end=8 下载视频id为BV1oK411L7au的系列教程,从第4开始,到第8个结束,如果只有一个的话,start和end设为1即可。

可以不断地往队列里面添加下载链接。
主要代码:
# @Time : 2019/1/28 14:19
# @File : youtube_downloader.py

import logging
import os
import subprocess
import datetime
import sqlite3
import time
from config import YOU_GET_PATH,MINS
CMD = 'python {} {}'
filename = 'url.txt'


class SQLite():
def __init__(self):
self.conn = sqlite3.connect('history.db')
self.cursor = self.conn.cursor()
self.create_table()

def create_table(self):
create_sql = 'create table if not exists tb_download (url varchar(100),status tinyint,crawltime datetime)'
create_record_tb = 'create table if not exists tb_record (idx varchar(100) PRIMARY KEY,start tinyint,end tinyint,status tinyint)'
self.cursor.execute(create_record_tb)
self.conn.commit()
self.cursor.execute(create_sql)
self.conn.commit()

def exists(self,url):
querySet = 'select * from tb_download where url = ? and status = 1'
self.cursor.execute(querySet,(url,))
ret = self.cursor.fetchone()
return True if ret else False

def insert_history(self,url,status):
query = 'select * from tb_download where url=?'
self.cursor.execute(query,(url,))
ret = self.cursor.fetchone()
current = datetime.datetime.now()

if ret:
insert_sql='update tb_download set status=?,crawltime=? where url = ?'
args=(status,status,current,url)
else:
insert_sql = 'insert into tb_download values(?,?,?)'
args=(url,status,current)

try:
self.cursor.execute(insert_sql,args)
except:
self.conn.rollback()
return False
else:
self.conn.commit()
return True

def get(self):
sql = 'select idx,start,end from tb_record where status=0'
self.cursor.execute(sql)
ret= self.cursor.fetchone()
return ret

def set(self,idx):
print('set status =1')
sql='update tb_record set status=1 where idx=?'
self.cursor.execute(sql,(idx,))
self.conn.commit()

def llogger(filename):
logger = logging.getLogger(filename) # 不加名称设置root logger

logger.setLevel(logging.DEBUG) # 设置输出级别

formatter = logging.Formatter(
'[%(asctime)s][%(filename)s][line: %(lineno)d]\[%(levelname)s] ## %(message)s)',
datefmt='%Y-%m-%d %H:%M:%S')

# 使用FileHandler输出到文件
prefix = os.path.splitext(filename)[0]
fh = logging.FileHandler(prefix + '.log')
fh.setLevel(logging.DEBUG)
fh.setFormatter(formatter)

# 使用StreamHandler输出到屏幕
ch = logging.StreamHandler()
ch.setLevel(logging.DEBUG)
ch.setFormatter(formatter)

# 添加两个Handler
logger.addHandler(ch)
logger.addHandler(fh)
return logger


logger = llogger('download.log')
sql_obj = SQLite()

def run():
while 1:
result = sql_obj.get()
print(result)
if result:
idx=result[0]
start=result[1]
end=result[2]
try:
download_bilibili(idx,start,end)
except:
pass
else:
sql_obj.set(idx)
else:
time.sleep(MINS*60)

def download_bilibili(id,start_page,total_page):
global doc

bilibili_url = 'https://www.bilibili.com/video/{}?p={}'
for i in range(start_page, total_page+1):

next_url = bilibili_url.format(id, i)
if sql_obj.exists(next_url):
print('have download')
continue

try:
command = CMD.format(YOU_GET_PATH, next_url)
p = subprocess.Popen(command, stdout=subprocess.PIPE, stderr=subprocess.PIPE,
shell=True)

output, error = p.communicate()

except Exception as e:
print('has execption')
sql_obj.insert_history(next_url,status=0)
logger.error(e)
continue
else:
output_str = output.decode()
if len(output_str) == 0:
sql_obj.insert_history(next_url,status=0)
logger.info('下载失败')
continue

logger.info('{} has been downloaded !'.format(next_url))
sql_obj.insert_history(next_url,status=1)

run()

  查看全部
B站上不少优秀的学习资源,下载到本地观看,便于快进,多倍速。 也可以放到平板,手机,在没有网络,或者网络条件不佳的环境下观看。
 

 
使用python实现
https://github.com/Rockyzsu/bilibili
 
B站视频下载
自动批量下载B站一个系列的视频

下载某个UP主的所有视频

使用:

下载you-get库,git clone https://github.com/soimort/you-get.git 复制其本地路径,比如/root/you-get/you-get

初次运行,删除history.db 文件, 修改配置文件config.py

START=1 # 下载系列视频的 第一个
END=1 # 下载系列视频的最后一个 , 比如一个系列教程有30个视频, start=5 ,end = 20 下载从第5个到第20个
ID='BV1oK411L7au' # 视频的ID
YOU_GET_PATH='/home/xda/othergit/you-get/you-get' # 你的you-get路径
MINS=1 # 每次循环等待1分钟
user_id = '518973111' # UP主的ID
total_page = 3 # up主的视频的页数
执行 python downloader.py ,进行下载循环

python people.py ,把某个up主的视频链接加入到待下载队列

python add_data.py --id=BV1oK411L7au --start=4 --end=8 下载视频id为BV1oK411L7au的系列教程,从第4开始,到第8个结束,如果只有一个的话,start和end设为1即可。

可以不断地往队列里面添加下载链接。

主要代码:
# @Time : 2019/1/28 14:19
# @File : youtube_downloader.py

import logging
import os
import subprocess
import datetime
import sqlite3
import time
from config import YOU_GET_PATH,MINS
CMD = 'python {} {}'
filename = 'url.txt'


class SQLite():
def __init__(self):
self.conn = sqlite3.connect('history.db')
self.cursor = self.conn.cursor()
self.create_table()

def create_table(self):
create_sql = 'create table if not exists tb_download (url varchar(100),status tinyint,crawltime datetime)'
create_record_tb = 'create table if not exists tb_record (idx varchar(100) PRIMARY KEY,start tinyint,end tinyint,status tinyint)'
self.cursor.execute(create_record_tb)
self.conn.commit()
self.cursor.execute(create_sql)
self.conn.commit()

def exists(self,url):
querySet = 'select * from tb_download where url = ? and status = 1'
self.cursor.execute(querySet,(url,))
ret = self.cursor.fetchone()
return True if ret else False

def insert_history(self,url,status):
query = 'select * from tb_download where url=?'
self.cursor.execute(query,(url,))
ret = self.cursor.fetchone()
current = datetime.datetime.now()

if ret:
insert_sql='update tb_download set status=?,crawltime=? where url = ?'
args=(status,status,current,url)
else:
insert_sql = 'insert into tb_download values(?,?,?)'
args=(url,status,current)

try:
self.cursor.execute(insert_sql,args)
except:
self.conn.rollback()
return False
else:
self.conn.commit()
return True

def get(self):
sql = 'select idx,start,end from tb_record where status=0'
self.cursor.execute(sql)
ret= self.cursor.fetchone()
return ret

def set(self,idx):
print('set status =1')
sql='update tb_record set status=1 where idx=?'
self.cursor.execute(sql,(idx,))
self.conn.commit()

def llogger(filename):
logger = logging.getLogger(filename) # 不加名称设置root logger

logger.setLevel(logging.DEBUG) # 设置输出级别

formatter = logging.Formatter(
'[%(asctime)s][%(filename)s][line: %(lineno)d]\[%(levelname)s] ## %(message)s)',
datefmt='%Y-%m-%d %H:%M:%S')

# 使用FileHandler输出到文件
prefix = os.path.splitext(filename)[0]
fh = logging.FileHandler(prefix + '.log')
fh.setLevel(logging.DEBUG)
fh.setFormatter(formatter)

# 使用StreamHandler输出到屏幕
ch = logging.StreamHandler()
ch.setLevel(logging.DEBUG)
ch.setFormatter(formatter)

# 添加两个Handler
logger.addHandler(ch)
logger.addHandler(fh)
return logger


logger = llogger('download.log')
sql_obj = SQLite()

def run():
while 1:
result = sql_obj.get()
print(result)
if result:
idx=result[0]
start=result[1]
end=result[2]
try:
download_bilibili(idx,start,end)
except:
pass
else:
sql_obj.set(idx)
else:
time.sleep(MINS*60)

def download_bilibili(id,start_page,total_page):
global doc

bilibili_url = 'https://www.bilibili.com/video/{}?p={}'
for i in range(start_page, total_page+1):

next_url = bilibili_url.format(id, i)
if sql_obj.exists(next_url):
print('have download')
continue

try:
command = CMD.format(YOU_GET_PATH, next_url)
p = subprocess.Popen(command, stdout=subprocess.PIPE, stderr=subprocess.PIPE,
shell=True)

output, error = p.communicate()

except Exception as e:
print('has execption')
sql_obj.insert_history(next_url,status=0)
logger.error(e)
continue
else:
output_str = output.decode()
if len(output_str) == 0:
sql_obj.insert_history(next_url,status=0)
logger.info('下载失败')
continue

logger.info('{} has been downloaded !'.format(next_url))
sql_obj.insert_history(next_url,status=1)

run()

 

python3的map是迭代器,不用for循环或者next触发是不会执行的

python李魔佛 发表了文章 • 0 个评论 • 167 次浏览 • 2022-05-21 17:59 • 来自相关话题

最近刚好有位群友咨询,他写的代码如下:
def update_data(id,start,end):
status=0
conn = sqlite3.connect('history.db')
cursor = conn.cursor()
insert_sql ='insert into tb_record values(?,?,?,?)'

try:
cursor.execute(insert_sql,(id,start,end,status))
except Exception as e:
print(e)
print('Error')
else:
conn.commit()
print("successfully insert") 
bv_list = []
for i in range(1, total_page + 1):
bv_list.extend(visit(i))
print(bv_list)
map(lambda x:update_data(x,1,1),bv_list)
作用很简单,就是拿到列表后用map放入到sqlite里面。
但是上面的代码并不起作用。
因为map只是定义了一个迭代器,并没有被触发。
 
可以加一个list(map(lambda x:update_data(x,1,1),bv_list))
这样就可以执行了。 查看全部
最近刚好有位群友咨询,他写的代码如下:
def update_data(id,start,end):
status=0
conn = sqlite3.connect('history.db')
cursor = conn.cursor()
insert_sql ='insert into tb_record values(?,?,?,?)'

try:
cursor.execute(insert_sql,(id,start,end,status))
except Exception as e:
print(e)
print('Error')
else:
conn.commit()
print("successfully insert")
 
bv_list = []
for i in range(1, total_page + 1):
bv_list.extend(visit(i))
print(bv_list)
map(lambda x:update_data(x,1,1),bv_list)

作用很简单,就是拿到列表后用map放入到sqlite里面。
但是上面的代码并不起作用。
因为map只是定义了一个迭代器,并没有被触发。
 
可以加一个list(map(lambda x:update_data(x,1,1),bv_list))
这样就可以执行了。

dataframe如何 遍历所有的列?

python李魔佛 发表了文章 • 0 个评论 • 290 次浏览 • 2022-05-21 02:16 • 来自相关话题

如果遍历行,我们经常会使用df.iterrows(), 而列呢?
可以使用df.items()
 Python pandas.DataFrame.items用法及代码示例
用法:
DataFrame.items()
迭代(列名,系列)对。

遍历 DataFrame 列,返回一个包含列名和内容的元组作为一个系列。

生成(Yield):
label:对象
被迭代的 DataFrame 的列名。

content:Series
属于每个标签的列条目,作为一个系列。

例子:
>>> df = pd.DataFrame({'species':['bear', 'bear', 'marsupial'],
... 'population':[1864, 22000, 80000]},
... index=['panda', 'polar', 'koala'])
>>> df
species population
panda bear 1864
polar bear 22000
koala marsupial 80000
>>> for label, content in df.items():
... print(f'label:{label}')
... print(f'content:{content}', sep='\n')
...
label:species
content:
panda bear
polar bear
koala marsupial
Name:species, dtype:object
label:population
content:
panda 1864
polar 22000
koala 80000
Name:population, dtype:int64 查看全部
如果遍历行,我们经常会使用df.iterrows(), 而列呢?
可以使用df.items()
 
Python pandas.DataFrame.items用法及代码示例
用法:
DataFrame.items()
迭代(列名,系列)对。

遍历 DataFrame 列,返回一个包含列名和内容的元组作为一个系列。

生成(Yield):
label:对象
被迭代的 DataFrame 的列名。

content:Series
属于每个标签的列条目,作为一个系列。

例子:
>>> df = pd.DataFrame({'species':['bear', 'bear', 'marsupial'],
... 'population':[1864, 22000, 80000]},
... index=['panda', 'polar', 'koala'])
>>> df
species population
panda bear 1864
polar bear 22000
koala marsupial 80000
>>> for label, content in df.items():
... print(f'label:{label}')
... print(f'content:{content}', sep='\n')
...
label:species
content:
panda bear
polar bear
koala marsupial
Name:species, dtype:object
label:population
content:
panda 1864
polar 22000
koala 80000
Name:population, dtype:int64

python3 安装demjson 报错 use_2to3 is invalid

python李魔佛 发表了文章 • 0 个评论 • 541 次浏览 • 2022-04-18 20:19 • 来自相关话题

ooking in indexes: https://pypi.douban.com/simple
Collecting demjson==2.2.4
Downloading https://pypi.doubanio.com/pack ... ar.gz (131 kB)
|████████████████████████████████| 131 kB 985 kB/s
ERROR: Command errored out with exit status 1:
command: /root/miniconda3/envs/py37/bin/python -c 'import io, os, sys, setuptools, tokenize; sys.argv[0] = '"'"'/tmp/pip-install-pj0ajcpl/demjson_86a1bab6643c4ed7a7b0c6bb6d3a43a2/setup.py'"'"'; __file__='"'"'/tmp/pip-install-pj0ajcpl/demjson_86a1bab6643c4ed7a7b0c6bb6d3a43a2/setup.py'"'"';f = getattr(tokenize, '"'"'open'"'"', open)(__file__) if os.path.exists(__file__) else io.StringIO('"'"'from setuptools import setup; setup()'"'"');code = f.read().replace('"'"'\r\n'"'"', '"'"'\n'"'"');f.close();exec(compile(code, __file__, '"'"'exec'"'"'))' egg_info --egg-base /tmp/pip-pip-egg-info-7ve4tu87
cwd: /tmp/pip-install-pj0ajcpl/demjson_86a1bab6643c4ed7a7b0c6bb6d3a43a2/
Complete output (3 lines):
/root/miniconda3/envs/py37/lib/python3.7/site-packages/setuptools/dist.py:760: UserWarning: Usage of dash-separated 'index-url' will not be supported in future versions. Please use the underscore name 'index_url' instead
% (opt, underscore_opt)
error in demjson setup command: use_2to3 is invalid.

 记录一下解决办法:
setuptools 降级:
 
pip install --upgrade setuptools==57.5.0
 
然后再pip install demjson 即可
 
如果担心setuptools 修改到系统的其他库,可以创建一个虚拟环境。
然后在虚拟环境里面对setuptools 降级,再安装demjson 
 
来个养眼图: 查看全部
ooking in indexes: https://pypi.douban.com/simple
Collecting demjson==2.2.4
Downloading https://pypi.doubanio.com/pack ... ar.gz (131 kB)
|████████████████████████████████| 131 kB 985 kB/s
ERROR: Command errored out with exit status 1:
command: /root/miniconda3/envs/py37/bin/python -c 'import io, os, sys, setuptools, tokenize; sys.argv[0] = '"'"'/tmp/pip-install-pj0ajcpl/demjson_86a1bab6643c4ed7a7b0c6bb6d3a43a2/setup.py'"'"'; __file__='"'"'/tmp/pip-install-pj0ajcpl/demjson_86a1bab6643c4ed7a7b0c6bb6d3a43a2/setup.py'"'"';f = getattr(tokenize, '"'"'open'"'"', open)(__file__) if os.path.exists(__file__) else io.StringIO('"'"'from setuptools import setup; setup()'"'"');code = f.read().replace('"'"'\r\n'"'"', '"'"'\n'"'"');f.close();exec(compile(code, __file__, '"'"'exec'"'"'))' egg_info --egg-base /tmp/pip-pip-egg-info-7ve4tu87
cwd: /tmp/pip-install-pj0ajcpl/demjson_86a1bab6643c4ed7a7b0c6bb6d3a43a2/
Complete output (3 lines):
/root/miniconda3/envs/py37/lib/python3.7/site-packages/setuptools/dist.py:760: UserWarning: Usage of dash-separated 'index-url' will not be supported in future versions. Please use the underscore name 'index_url' instead
% (opt, underscore_opt)
error in demjson setup command: use_2to3 is invalid.

 记录一下解决办法:
setuptools 降级:
 
pip install --upgrade setuptools==57.5.0
 
然后再pip install demjson 即可
 
如果担心setuptools 修改到系统的其他库,可以创建一个虚拟环境。
然后在虚拟环境里面对setuptools 降级,再安装demjson 
 
来个养眼图:

centos/ubuntu linux 下 python matplotlib 绘图中文标题乱码 解决

Linux李魔佛 发表了文章 • 0 个评论 • 290 次浏览 • 2022-04-15 22:23 • 来自相关话题

出现下图的错误:





 测试代码:import matplotlib as mpl
import matplotlib.pyplot as plt
from matplotlib import rcParams


plt.rcParams['font.family'] = 'sans-serif'
plt.rcParams['font.sans-serif'] = ['simhei'] # 指定默认字体
plt.rcParams['axes.unicode_minus'] = False

plt.title(r'宋体 Times New Roman')
plt.axis('off')
plt.savefig('usestix.png') 
每次换机器都遇到这个问题,所以记录下来,以免日后又在折腾
首先找到一个ttf格式的中文字体,windows系统的可以在C盘的system里面找找。
我用的是微软雅黑。msyh.ttf , 需要的可以到公众号里面取。 后台回复: msyh.ttf 即可
或者simhei.ttfcd /usr/share/fonts/
sudo mkdir chinese
sudo cp simhei.ttf /usr/share/fonts/chinese/
sudo chmod -R 755 /usr/share/fonts/chinese
更新缓存mkfontscale
mkfontdir
fc-cache


注意,如果你的系统没有上述命令,需要使用yum install mkfontscale mkfontdir c-cache安装
 
然后接下来这一步,很重要。
 
清理缓存,不然不会生效!!!!rm -rf ~/.cache/matplotlib


 
matplotlib:
同时把这个msyh.ttf的文件复制到matplotlib库下的文件夹:
~python3.9/site-packages/matplotlib/mpl-data/fonts/ttf/ 
下面
 
然后运行程序,搞定~





 
公众号:可转债量化分析

后台回复 : msyh.ttf 或者 simhei.ttf 即可获取字体文件。 查看全部
出现下图的错误:

20220415_688533_上声电子_zdt.png

 测试代码:
import matplotlib as mpl
import matplotlib.pyplot as plt
from matplotlib import rcParams


plt.rcParams['font.family'] = 'sans-serif'
plt.rcParams['font.sans-serif'] = ['simhei'] # 指定默认字体
plt.rcParams['axes.unicode_minus'] = False

plt.title(r'宋体 Times New Roman')
plt.axis('off')
plt.savefig('usestix.png')
 
每次换机器都遇到这个问题,所以记录下来,以免日后又在折腾
首先找到一个ttf格式的中文字体,windows系统的可以在C盘的system里面找找。
我用的是微软雅黑。msyh.ttf , 需要的可以到公众号里面取。 后台回复: msyh.ttf 即可
或者simhei.ttf
cd /usr/share/fonts/
sudo mkdir chinese
sudo cp simhei.ttf /usr/share/fonts/chinese/
sudo chmod -R 755 /usr/share/fonts/chinese

更新缓存
mkfontscale
mkfontdir
fc-cache


注意,如果你的系统没有上述命令,需要使用yum install mkfontscale mkfontdir c-cache安装
 
然后接下来这一步,很重要。
 
清理缓存,不然不会生效!!!!
rm -rf ~/.cache/matplotlib


 
matplotlib:
同时把这个msyh.ttf的文件复制到matplotlib库下的文件夹:
~python3.9/site-packages/matplotlib/mpl-data/fonts/ttf/ 
下面
 
然后运行程序,搞定~

20220415_688533_上声电子_zdt_(2).png

 
公众号:可转债量化分析

后台回复 : msyh.ttf 或者 simhei.ttf 即可获取字体文件。

知识星球获取文章链接与数据

python爬虫李魔佛 发表了文章 • 0 个评论 • 314 次浏览 • 2022-03-21 20:15 • 来自相关话题

 
既然官方不提供这个功能,只能自己使用爬虫手段获取了,额。

 
既然官方不提供这个功能,只能自己使用爬虫手段获取了,额。

python AES 加密 windows和linux平台的不同

python李魔佛 发表了文章 • 0 个评论 • 350 次浏览 • 2022-03-19 11:18 • 来自相关话题

同样一段AES加密的代码,放到了ubuntu可以正常使用,而在windows却报错。
实际两个平台使用pip install安装的aes库不一样。
 
windows报错
File "C:\anaconda\lib\site-packages\Crypto\Cipher\__init__.py", line 77, in _create_cipher
raise TypeError("IV is not meaningful for the ECB mode")
TypeError: IV is not meaningful for the ECB mode

只需要把AES.new() 的参数里面的iv给去掉就可以了。
 

  查看全部
同样一段AES加密的代码,放到了ubuntu可以正常使用,而在windows却报错。
实际两个平台使用pip install安装的aes库不一样。
 
windows报错
  File "C:\anaconda\lib\site-packages\Crypto\Cipher\__init__.py", line 77, in _create_cipher
raise TypeError("IV is not meaningful for the ECB mode")
TypeError: IV is not meaningful for the ECB mode

只需要把AES.new() 的参数里面的iv给去掉就可以了。
 

 

不是所有的bytes都可以转换为string

python李魔佛 发表了文章 • 0 个评论 • 494 次浏览 • 2022-01-14 14:56 • 来自相关话题

byte转为string
b.decode('utf8')
 
如果报错:UnicodeDecodeError: 'utf-8' codec can't decode byte 0xe4 in position 1: invalid continuation byte
说明字节无法字节转为string, 





 
上面的字节是可以正常decode为utf8
 
而改下字节数据





 
所以你试下decode下面的字节:c=b'\x1e\xe4\xd5\x97\x9a#\x99kC\xadD\x7f\x9a\xc2G\x92'
是无法解析的。
 
这个是没有办法的,如果要硬刚 。
可以加入参数errors = ‘replace’
 b.decode('utf8',errors='replace')
这样就不会报错。但是这也只是输出的乱码。
 
正确的姿势是要看看你的字节的最原始编码格式。如果是gbk,那么就应该使用b.decode('gbk')
 
  查看全部
byte转为string
b.decode('utf8')
 
如果报错:
UnicodeDecodeError: 'utf-8' codec can't decode byte 0xe4 in position 1: invalid continuation byte

说明字节无法字节转为string, 

Shutter_2022-01-14-14:49:54.png

 
上面的字节是可以正常decode为utf8
 
而改下字节数据

Shutter_2022-01-14-14:53:40.png

 
所以你试下decode下面的字节:
c=b'\x1e\xe4\xd5\x97\x9a#\x99kC\xadD\x7f\x9a\xc2G\x92'

是无法解析的。
 
这个是没有办法的,如果要硬刚 。
可以加入参数errors = ‘replace’
 
b.decode('utf8',errors='replace')

这样就不会报错。但是这也只是输出的乱码。
 
正确的姿势是要看看你的字节的最原始编码格式。如果是gbk,那么就应该使用b.decode('gbk')
 
 

目前能用的 可以下载youtube 油管视频的方法 合集 亲测

网络李魔佛 发表了文章 • 0 个评论 • 1768 次浏览 • 2021-11-08 13:03 • 来自相关话题

最近的saveform网站下架了油管视频下载的功能。
 
Subject: Discontinuation of Service in the United States

April 16, 2020

Dear Savefrom User:

As you may have heard, our industry has been under strenuous attacks by certain US copyright holders. Because of these attacks, it has become financially impractical for Savefrom to continue to provide services in the United States.

Accordingly, Savefrom will be terminating its services in the United States as of April 28, 2020.

We thank you for your past loyalty and patronage and wish you health and safety during the present health crisis and beyond.

Very truly yours,

Savefrom所以这个网页下载的方法失效了。
 
还有其他 可以使用的方法:
 
方法一   you-get
这是一个python库。
在github下载下来
默认github下载是很卡,且断断续续,具体原因,你懂的。
所以笔者提供一个镜像下载,代码也是最新的。
git clone https://hub.fastgit.org/soimort/you-get.git
运行上面命令即可。
 
然后下载过程就是切换的CMD命令行:
python you-get --debug https://www.youtube.com/watch?v=59a-XNPngrw
 
静静等待即可下载。
 

  查看全部
最近的saveform网站下架了油管视频下载的功能。
 
Subject: Discontinuation of Service in the United States

April 16, 2020

Dear Savefrom User:

As you may have heard, our industry has been under strenuous attacks by certain US copyright holders. Because of these attacks, it has become financially impractical for Savefrom to continue to provide services in the United States.

Accordingly, Savefrom will be terminating its services in the United States as of April 28, 2020.

We thank you for your past loyalty and patronage and wish you health and safety during the present health crisis and beyond.

Very truly yours,

Savefrom
所以这个网页下载的方法失效了。
 
还有其他 可以使用的方法:
 
方法一   you-get
这是一个python库。
在github下载下来
默认github下载是很卡,且断断续续,具体原因,你懂的。
所以笔者提供一个镜像下载,代码也是最新的。
git clone https://hub.fastgit.org/soimort/you-get.git
运行上面命令即可。
 
然后下载过程就是切换的CMD命令行:
python you-get --debug https://www.youtube.com/watch?v=59a-XNPngrw
 
静静等待即可下载。
 

 

通达信公式转python代码

股票量化大师 发表了文章 • 0 个评论 • 1246 次浏览 • 2021-08-12 18:10 • 来自相关话题

本人目前正在做这个转换工作。并定期更新发布在这里。有兴趣的可以参与。
本人目前正在做这个转换工作。并定期更新发布在这里。有兴趣的可以参与。

如何使用控制台将动态加载数据刷新出来啊????

python低调的哥哥 回复了问题 • 2 人关注 • 1 个回复 • 828 次浏览 • 2021-08-11 02:06 • 来自相关话题

python 上传文件夹内图片到七牛,同时加入批量删除,单个删除

python李魔佛 发表了文章 • 0 个评论 • 756 次浏览 • 2021-08-07 17:08 • 来自相关话题

先注册好七牛的账户,那都AK和SK两个key
 
然后把key写入到环境变量或者写到下面的python文件里面from qiniu import Auth, put_file,BucketManager,build_batch_delete
import os
import fire

access_key = os.getenv('qiniu_access_key')
secret_key = os.getenv('qiniu_secret_key')

bucket_name = '' # 你的空间名称

HOST ='[url]http://xximg.xxx.com/{}'[/url] # 可以不用填

TEMPLATE = '\n![{}]({})\n\n\n'

def upload(file,category=''):

    #构建鉴权对象
    q = Auth(access_key, secret_key)
    #要上传的空间

    #上传后保存的文件名
    key = category +'/' + os.path.split(file)[1]
    #生成上传 Token,可以指定过期时间等
    token = q.upload_token(bucket_name, key) # 永不过期

    #要上传文件的本地路径
    ret, info = put_file(token, key, file, version='v1') 
    print(ret)
    print(info)
    return HOST.format(ret['key'])

def bulk_upload(path,category=''):
    with open('qiniu_image.md','a+') as fp:
        for file in os.listdir(path):
            full_path = os.path.join(path,file)
            if os.path.isfile(full_path):

                host_url = upload(full_path,category)
                fp.write(TEMPLATE.format(host_url,host_url))


def get_file_info(prefix,limit = 10):
    q = Auth(access_key, secret_key)
    bucket = BucketManager(q)

    delimiter = None
    marker = None
    ret, eof, info = bucket.list(bucket_name, prefix, marker, limit, delimiter)

    # assert len(ret.get('items')) is not None
    url_list=[]
    for item in ret.get('items',):
        url_list.append(item['key'])
    # print(url_list)
    # print(len(url_list))
    return url_list,len(url_list)

def bulk_delete(prefix,limit=None):
    url_list,lens = get_file_info(prefix,limit=limit)
    q = Auth(access_key, secret_key)
    bucket = BucketManager(q)
    ops = build_batch_delete(bucket_name, url_list)
    ret, info = bucket.batch(ops)
    print(info)
    print(ret)

def delete_one(key):
    q = Auth(access_key, secret_key)
    #初始化BucketManager
    bucket = BucketManager(q)
    #你要测试的空间, 并且这个key在你空间中存在
    # key = 'python-logo.png'
    #删除bucket_name 中的文件 key
    ret, info = bucket.delete(bucket_name, key)
    print(info)
    print(ret)
    # assert ret == {}


def bulk_delete_ones(prefix):
    url_list,lens = get_file_info(prefix,limit=10)
    for url in url_list:
        delete_one(url)
        # print(url)

def main(path,category):

    if os.path.isdir(path):

         bulk_upload(path,category)
    elif os.path.isfile(path):
         upload(path,category)
    else:
         raise ValueError('文件不存在')

    get_file_info()
    bulk_delete('resource')
    bulk_delete_ones('resource')
    delete_one('resource/data_beauty.png')

if __name__ == '__main__':
    fire.Fire(main)



然后运行:
python main.py --path='C:\Photo' --category='person'
运行后会上传到七牛的虚拟目录 person目录下
 
如果要删除,bulk_delete批量删除某个前缀或者文件夹的  查看全部
先注册好七牛的账户,那都AK和SK两个key
 
然后把key写入到环境变量或者写到下面的python文件里面
from qiniu import Auth, put_file,BucketManager,build_batch_delete
import os
import fire

access_key = os.getenv('qiniu_access_key')
secret_key = os.getenv('qiniu_secret_key')

bucket_name = '' # 你的空间名称

HOST ='[url]http://xximg.xxx.com/{}'[/url] # 可以不用填

TEMPLATE = '\n![{}]({})\n\n\n'

def upload(file,category=''):

    #构建鉴权对象
    q = Auth(access_key, secret_key)
    #要上传的空间

    #上传后保存的文件名
    key = category +'/' + os.path.split(file)[1]
    #生成上传 Token,可以指定过期时间等
    token = q.upload_token(bucket_name, key) # 永不过期

    #要上传文件的本地路径
    ret, info = put_file(token, key, file, version='v1') 
    print(ret)
    print(info)
    return HOST.format(ret['key'])

def bulk_upload(path,category=''):
    with open('qiniu_image.md','a+') as fp:
        for file in os.listdir(path):
            full_path = os.path.join(path,file)
            if os.path.isfile(full_path):

                host_url = upload(full_path,category)
                fp.write(TEMPLATE.format(host_url,host_url))


def get_file_info(prefix,limit = 10):
    q = Auth(access_key, secret_key)
    bucket = BucketManager(q)

    delimiter = None
    marker = None
    ret, eof, info = bucket.list(bucket_name, prefix, marker, limit, delimiter)

    # assert len(ret.get('items')) is not None
    url_list=[]
    for item in ret.get('items',):
        url_list.append(item['key'])
    # print(url_list)
    # print(len(url_list))
    return url_list,len(url_list)

def bulk_delete(prefix,limit=None):
    url_list,lens = get_file_info(prefix,limit=limit)
    q = Auth(access_key, secret_key)
    bucket = BucketManager(q)
    ops = build_batch_delete(bucket_name, url_list)
    ret, info = bucket.batch(ops)
    print(info)
    print(ret)

def delete_one(key):
    q = Auth(access_key, secret_key)
    #初始化BucketManager
    bucket = BucketManager(q)
    #你要测试的空间, 并且这个key在你空间中存在
    # key = 'python-logo.png'
    #删除bucket_name 中的文件 key
    ret, info = bucket.delete(bucket_name, key)
    print(info)
    print(ret)
    # assert ret == {}


def bulk_delete_ones(prefix):
    url_list,lens = get_file_info(prefix,limit=10)
    for url in url_list:
        delete_one(url)
        # print(url)

def main(path,category):

    if os.path.isdir(path):

         bulk_upload(path,category)
    elif os.path.isfile(path):
         upload(path,category)
    else:
         raise ValueError('文件不存在')

    get_file_info()
    bulk_delete('resource')
    bulk_delete_ones('resource')
    delete_one('resource/data_beauty.png')

if __name__ == '__main__':
    fire.Fire(main)



然后运行:
python main.py --path='C:\Photo' --category='person'
运行后会上传到七牛的虚拟目录 person目录下
 
如果要删除,bulk_delete批量删除某个前缀或者文件夹的 

为什么登录成功但是爬取不到其他数据

python爬虫低调的哥哥 回复了问题 • 2 人关注 • 1 个回复 • 854 次浏览 • 2021-08-04 01:17 • 来自相关话题

优矿接口的日期定义真让人蛋疼

股票李魔佛 发表了文章 • 0 个评论 • 761 次浏览 • 2021-07-17 20:24 • 来自相关话题

日期查询格式是 YYYYMMDD的.
然后接口查询结果返回的是YYYY-MM-DD
难道统一一下这么难吗?
日期查询格式是 YYYYMMDD的.
然后接口查询结果返回的是YYYY-MM-DD
难道统一一下这么难吗?

开通Ptrade Python量化接口 国金证券/国盛证券

股票绫波丽 发表了文章 • 0 个评论 • 6704 次浏览 • 2021-07-06 08:40 • 来自相关话题

目前不少在A股做量化的大部分使用tushare,优矿,米筐,聚宽等等,无论教程,还是实际操作,基本没见有教怎么用程序下单,实盘交易的。

而退而求其次使用按键精灵,模拟点击交易软件进行点击下单,非常不稳定,无法判断下单后是否成交,也无法实时获取行情数据。如果使用tushare或者新浪接口数据,扫描一次全市场的行情用时很久且不稳定,等扫描结束,再下单,此时价格可能已经是几分钟前的了,且此类接口调用次数多是会被封IP的。

笔者使用的是券商提供的量化软件:Ptrade。是恒生电子研发的提供给机构使用的程序化交易软件。提供策略回测,下单API接口,实时行情获取,并且使用的开发语言python,易于上手。

策略回测与实盘交易




研究页面

研究页面,熟悉python jupyter notebook的朋友对这个界面肯定很熟悉。

研究的页面实际就运行你逐行输出调试程序,了解每个函数的具体使用,或者你策略的中途结果调试。





 
回测策略

实际代码需要在回测策略里面写,写完后确定无误,就可以放在仿真环境下真实运行。如果你运行得到的结果很满意,那么就可以直接部署到实盘服务器上。实盘服务器是在券商那边,不需要个人购买服务器,也不需要本地开着这个Ptrade,就是说不需要在个人电脑上一直开着跑,你的最终代码和程序是在券商服务器上部署与运行,除非有报错异常停止,不然在你不暂停或者停止的前提下,可以一直运行下去。












条件满足后下单

可视化量化

同时也提供一些常见的现成的量化策略,选中后只要鼠标点点点也能够自动化跑这些策略了,当然里面很多参数都可以用鼠标点点点修改。





 
接口文档也非常详细:




 
一些常见策略代码:

集合竞价追涨停策略def initialize(context):
# 初始化此策略
# 设置我们要操作的股票池, 这里我们只操作一支股票
g.security = '600570.SS'
set_universe(g.security)
#每天9:23分运行集合竞价处理函数
run_daily(context, aggregate_auction_func, time='9:23')

def aggregate_auction_func(context):
stock = g.security
#最新价
snapshot = get_snapshot(stock)
price = snapshot[stock]['last_px']
#涨停价
up_limit = snapshot[stock]['up_px']
#如果最新价不小于涨停价,买入
if float(price) >= float(up_limit):
order(g.security, 100, limit_price=up_limit)

def handle_data(context, data):
pass
双均线策略def initialize(context):
# 初始化此策略
# 设置我们要操作的股票池, 这里我们只操作一支股票
g.security = '600570.SS'
set_universe(g.security)
pass

#当五日均线高于十日均线时买入,当五日均线低于十日均线时卖出
def handle_data(context, data):
security = g.security

#得到十日历史价格
df = get_history(10, '1d', 'close', security, fq=None, include=False)

# 得到五日均线价格
ma5 = round(df['close'][-5:].mean(), 3)

# 得到十日均线价格
ma10 = round(df['close'][-10:].mean(), 3)

# 取得昨天收盘价
price = data[security]['close']

# 得到当前资金余额
cash = context.portfolio.cash

# 如果当前有余额,并且五日均线大于十日均线
if ma5 > ma10:
# 用所有 cash 买入股票
order_value(security, cash)
# 记录这次买入
log.info("Buying %s" % (security))

# 如果五日均线小于十日均线,并且目前有头寸
elif ma5 < ma10 and get_position(security).amount > 0:
# 全部卖出
order_target(security, 0)
# 记录这次卖出
log.info("Selling %s" % (security))
 
tick级别均线策略

通俗点就是按照秒级别进行操作。def initialize(context):
# 初始化此策略
# 设置我们要操作的股票池, 这里我们只操作一支股票
g.security = '600570.SS'
set_universe(g.security)
#每3秒运行一次主函数
run_interval(context, func, seconds=3)

#盘前准备历史数据
def before_trading_start(context, data):
history = get_history(10, '1d', 'close', g.security, fq='pre', include=False)
g.close_array = history['close'].values

#当五日均线高于十日均线时买入,当五日均线低于十日均线时卖出
def func(context):

stock = g.security

#获取最新价
snapshot = get_snapshot(stock)
price = snapshot[stock]['last_px']

# 得到五日均线价格
days = 5
ma5 = get_MA_day(stock, days, g.close_array[-4:], price)
# 得到十日均线价格
days = 10
ma10 = get_MA_day(stock, days, g.close_array[-9:], price)

# 得到当前资金余额
cash = context.portfolio.cash

# 如果当前有余额,并且五日均线大于十日均线
if ma5 > ma10:
# 用所有 cash 买入股票
order_value(stock, cash)
# 记录这次买入
log.info("Buying %s" % (stock))

# 如果五日均线小于十日均线,并且目前有头寸
elif ma5 < ma10 and get_position(stock).amount > 0:
# 全部卖出
order_target(stock, 0)
# 记录这次卖出
log.info("Selling %s" % (stock))

#计算实时均线函数
def get_MA_day(stock,days,close_array,current_price):
close_sum = close_array[-(days-1):].sum()
MA = (current_price + close_sum)/days
return MA

def handle_data(context, data):
pass
 
macd策略def f_expma(N,m,EXPMA1,price):
a = m/(N+1)
EXPMA2 = a * price + (1 - a)*EXPMA1
return EXPMA2 #2为后一天值

#定义macd函数,输入平滑系数参数、前一日值,输出当日值
def macd(N1,N2,N3,m,EXPMA12_1,EXPMA26_1,DEA1,price):
EXPMA12_2 = f_expma(N1,m,EXPMA12_1,price)
EXPMA26_2 = f_expma(N2,m,EXPMA26_1,price)
DIF2 = EXPMA12_2 - EXPMA26_2
a = m/(N3+1)
DEA2 = a * DIF2 + (1 - a)*DEA1
BAR2=2*(DIF2-DEA2)
return EXPMA12_2,EXPMA26_2,DIF2,DEA2,BAR2

def initialize(context):
global init_price
init_price = None
# 获取沪深300股票
g.security = get_index_stocks('000300.SS')
#g.security = ['600570.SS']
# 设置我们要操作的股票池, 这里我们只操作一支股票
set_universe(g.security)

def handle_data(context, data):
# 获取历史数据,这里只获取了2天的数据,如果希望最终MACD指标结果更准确最好是获取
# 从股票上市至今的所有历史数据,即增加获取的天数
close_price = get_history(2, '1d', field='close', security_list=g.security)
#如果是停牌不进行计算
for security in g.security:
if data[security].is_open >0:
global init_price,EXPMA12_1,EXPMA26_1,EXPMA12_2,EXPMA26_2,DIF1,DIF2,DEA1,DEA2
if init_price is None:
init_price = close_price[security].mean()#nan和N-1个数,mean为N-1个数的均值
EXPMA12_1 = init_price
EXPMA26_1 = init_price
DIF1 = init_price
DEA1 = init_price
# m用于计算平滑系数a=m/(N+1)
m = 2.0
#设定指数平滑基期数
N1 = 12
N2 = 26
N3 = 9
EXPMA12_2,EXPMA26_2,DIF2,DEA2,BAR2 = macd(N1,N2,N3,m,EXPMA12_1,EXPMA26_1,DEA1,close_price[security][-1])
# 取得当前价格
current_price = data[security].price
# 取得当前的现金
cash = context.portfolio.cash
# DIF、DEA均为正,DIF向上突破DEA,买入信号参考
if DIF2 > 0 and DEA2 > 0 and DIF1 < DEA1 and DIF2 > DEA2:
# 计算可以买多少只股票
number_of_shares = int(cash/current_price)
# 购买量大于0时,下单
if number_of_shares > 0:
# 以市单价买入股票,日回测时即是开盘价
order(security, +number_of_shares)
# 记录这次买入
log.info("Buying %s" % (security))
# DIF、DEA均为负,DIF向下突破DEA,卖出信号参考
elif DIF2 < 0 and DEA2 < 0 and DIF1 > DEA1 and DIF2 < DEA2 and get_position(security).amount > 0:
# 卖出所有股票,使这只股票的最终持有量为0
order_target(security, 0)
# 记录这次卖出
log.info("Selling %s" % (security))
# 将今日的值赋给全局变量作为下一次前一日的值
DEA1 = DEA2
DIF1 = DIF2
EXPMA12_1 = EXPMA12_2
EXPMA26_1 = EXPMA26_2
 
软件与交易接口开通条件:

开通该券商后,存入资金指定资金即可开通。开通后股票交易费率万一

本身券商的交易费率为股票万一,可转债沪百万分之五,深十万分之五,基金万0.5,非常厚道。
 
不太了解量化行业的可以了解下,不少面向机构的量化交易软件的佣金是万2.5的,且开户门槛高,基本是500W以上,比如华泰的matic量化的门槛是1千万元起步。

所以笔者还是很推荐目前该券商的量化交易接口。

需要开通咨询了解的朋友可以扫码联系:


开通券商账户后可以 可以先试用,再考虑是否开通量化接口权限 查看全部
目前不少在A股做量化的大部分使用tushare,优矿,米筐,聚宽等等,无论教程,还是实际操作,基本没见有教怎么用程序下单,实盘交易的。

而退而求其次使用按键精灵,模拟点击交易软件进行点击下单,非常不稳定,无法判断下单后是否成交,也无法实时获取行情数据。如果使用tushare或者新浪接口数据,扫描一次全市场的行情用时很久且不稳定,等扫描结束,再下单,此时价格可能已经是几分钟前的了,且此类接口调用次数多是会被封IP的。

笔者使用的是券商提供的量化软件:Ptrade。是恒生电子研发的提供给机构使用的程序化交易软件。提供策略回测,下单API接口,实时行情获取,并且使用的开发语言python,易于上手。

策略回测与实盘交易
UrD5TUnxZD.png

研究页面

研究页面,熟悉python jupyter notebook的朋友对这个界面肯定很熟悉。

研究的页面实际就运行你逐行输出调试程序,了解每个函数的具体使用,或者你策略的中途结果调试。

Hc8f4UtMfW.png

 
回测策略

实际代码需要在回测策略里面写,写完后确定无误,就可以放在仿真环境下真实运行。如果你运行得到的结果很满意,那么就可以直接部署到实盘服务器上。实盘服务器是在券商那边,不需要个人购买服务器,也不需要本地开着这个Ptrade,就是说不需要在个人电脑上一直开着跑,你的最终代码和程序是在券商服务器上部署与运行,除非有报错异常停止,不然在你不暂停或者停止的前提下,可以一直运行下去。

Pig1iRRQnP.png



25YjBEdOqa.png


条件满足后下单

可视化量化

同时也提供一些常见的现成的量化策略,选中后只要鼠标点点点也能够自动化跑这些策略了,当然里面很多参数都可以用鼠标点点点修改。

kTmn9iOXaS.png

 
接口文档也非常详细:
v4QFDpHNpd.png

 
一些常见策略代码:

集合竞价追涨停策略
def initialize(context):
# 初始化此策略
# 设置我们要操作的股票池, 这里我们只操作一支股票
g.security = '600570.SS'
set_universe(g.security)
#每天9:23分运行集合竞价处理函数
run_daily(context, aggregate_auction_func, time='9:23')

def aggregate_auction_func(context):
stock = g.security
#最新价
snapshot = get_snapshot(stock)
price = snapshot[stock]['last_px']
#涨停价
up_limit = snapshot[stock]['up_px']
#如果最新价不小于涨停价,买入
if float(price) >= float(up_limit):
order(g.security, 100, limit_price=up_limit)

def handle_data(context, data):
pass

双均线策略
def initialize(context):
# 初始化此策略
# 设置我们要操作的股票池, 这里我们只操作一支股票
g.security = '600570.SS'
set_universe(g.security)
pass

#当五日均线高于十日均线时买入,当五日均线低于十日均线时卖出
def handle_data(context, data):
security = g.security

#得到十日历史价格
df = get_history(10, '1d', 'close', security, fq=None, include=False)

# 得到五日均线价格
ma5 = round(df['close'][-5:].mean(), 3)

# 得到十日均线价格
ma10 = round(df['close'][-10:].mean(), 3)

# 取得昨天收盘价
price = data[security]['close']

# 得到当前资金余额
cash = context.portfolio.cash

# 如果当前有余额,并且五日均线大于十日均线
if ma5 > ma10:
# 用所有 cash 买入股票
order_value(security, cash)
# 记录这次买入
log.info("Buying %s" % (security))

# 如果五日均线小于十日均线,并且目前有头寸
elif ma5 < ma10 and get_position(security).amount > 0:
# 全部卖出
order_target(security, 0)
# 记录这次卖出
log.info("Selling %s" % (security))

 
tick级别均线策略

通俗点就是按照秒级别进行操作。
def initialize(context):
# 初始化此策略
# 设置我们要操作的股票池, 这里我们只操作一支股票
g.security = '600570.SS'
set_universe(g.security)
#每3秒运行一次主函数
run_interval(context, func, seconds=3)

#盘前准备历史数据
def before_trading_start(context, data):
history = get_history(10, '1d', 'close', g.security, fq='pre', include=False)
g.close_array = history['close'].values

#当五日均线高于十日均线时买入,当五日均线低于十日均线时卖出
def func(context):

stock = g.security

#获取最新价
snapshot = get_snapshot(stock)
price = snapshot[stock]['last_px']

# 得到五日均线价格
days = 5
ma5 = get_MA_day(stock, days, g.close_array[-4:], price)
# 得到十日均线价格
days = 10
ma10 = get_MA_day(stock, days, g.close_array[-9:], price)

# 得到当前资金余额
cash = context.portfolio.cash

# 如果当前有余额,并且五日均线大于十日均线
if ma5 > ma10:
# 用所有 cash 买入股票
order_value(stock, cash)
# 记录这次买入
log.info("Buying %s" % (stock))

# 如果五日均线小于十日均线,并且目前有头寸
elif ma5 < ma10 and get_position(stock).amount > 0:
# 全部卖出
order_target(stock, 0)
# 记录这次卖出
log.info("Selling %s" % (stock))

#计算实时均线函数
def get_MA_day(stock,days,close_array,current_price):
close_sum = close_array[-(days-1):].sum()
MA = (current_price + close_sum)/days
return MA

def handle_data(context, data):
pass

 
macd策略
def f_expma(N,m,EXPMA1,price):
a = m/(N+1)
EXPMA2 = a * price + (1 - a)*EXPMA1
return EXPMA2 #2为后一天值

#定义macd函数,输入平滑系数参数、前一日值,输出当日值
def macd(N1,N2,N3,m,EXPMA12_1,EXPMA26_1,DEA1,price):
EXPMA12_2 = f_expma(N1,m,EXPMA12_1,price)
EXPMA26_2 = f_expma(N2,m,EXPMA26_1,price)
DIF2 = EXPMA12_2 - EXPMA26_2
a = m/(N3+1)
DEA2 = a * DIF2 + (1 - a)*DEA1
BAR2=2*(DIF2-DEA2)
return EXPMA12_2,EXPMA26_2,DIF2,DEA2,BAR2

def initialize(context):
global init_price
init_price = None
# 获取沪深300股票
g.security = get_index_stocks('000300.SS')
#g.security = ['600570.SS']
# 设置我们要操作的股票池, 这里我们只操作一支股票
set_universe(g.security)

def handle_data(context, data):
# 获取历史数据,这里只获取了2天的数据,如果希望最终MACD指标结果更准确最好是获取
# 从股票上市至今的所有历史数据,即增加获取的天数
close_price = get_history(2, '1d', field='close', security_list=g.security)
#如果是停牌不进行计算
for security in g.security:
if data[security].is_open >0:
global init_price,EXPMA12_1,EXPMA26_1,EXPMA12_2,EXPMA26_2,DIF1,DIF2,DEA1,DEA2
if init_price is None:
init_price = close_price[security].mean()#nan和N-1个数,mean为N-1个数的均值
EXPMA12_1 = init_price
EXPMA26_1 = init_price
DIF1 = init_price
DEA1 = init_price
# m用于计算平滑系数a=m/(N+1)
m = 2.0
#设定指数平滑基期数
N1 = 12
N2 = 26
N3 = 9
EXPMA12_2,EXPMA26_2,DIF2,DEA2,BAR2 = macd(N1,N2,N3,m,EXPMA12_1,EXPMA26_1,DEA1,close_price[security][-1])
# 取得当前价格
current_price = data[security].price
# 取得当前的现金
cash = context.portfolio.cash
# DIF、DEA均为正,DIF向上突破DEA,买入信号参考
if DIF2 > 0 and DEA2 > 0 and DIF1 < DEA1 and DIF2 > DEA2:
# 计算可以买多少只股票
number_of_shares = int(cash/current_price)
# 购买量大于0时,下单
if number_of_shares > 0:
# 以市单价买入股票,日回测时即是开盘价
order(security, +number_of_shares)
# 记录这次买入
log.info("Buying %s" % (security))
# DIF、DEA均为负,DIF向下突破DEA,卖出信号参考
elif DIF2 < 0 and DEA2 < 0 and DIF1 > DEA1 and DIF2 < DEA2 and get_position(security).amount > 0:
# 卖出所有股票,使这只股票的最终持有量为0
order_target(security, 0)
# 记录这次卖出
log.info("Selling %s" % (security))
# 将今日的值赋给全局变量作为下一次前一日的值
DEA1 = DEA2
DIF1 = DIF2
EXPMA12_1 = EXPMA12_2
EXPMA26_1 = EXPMA26_2

 
软件与交易接口开通条件:

开通该券商后,存入资金指定资金即可开通。开通后股票交易费率万一

本身券商的交易费率为股票万一,可转债沪百万分之五,深十万分之五,基金万0.5,非常厚道。
 
不太了解量化行业的可以了解下,不少面向机构的量化交易软件的佣金是万2.5的,且开户门槛高,基本是500W以上,比如华泰的matic量化的门槛是1千万元起步。

所以笔者还是很推荐目前该券商的量化交易接口。

需要开通咨询了解的朋友可以扫码联系:


开通券商账户后可以 可以先试用,再考虑是否开通量化接口权限

B站批量下载某个UP主的所有视频

python李魔佛 发表了文章 • 0 个评论 • 247 次浏览 • 2022-05-21 18:48 • 来自相关话题

B站上不少优秀的学习资源,下载到本地观看,便于快进,多倍速。 也可以放到平板,手机,在没有网络,或者网络条件不佳的环境下观看。
 

 
使用python实现
https://github.com/Rockyzsu/bilibili
 
B站视频下载
自动批量下载B站一个系列的视频

下载某个UP主的所有视频

使用:

下载you-get库,git clone https://github.com/soimort/you-get.git 复制其本地路径,比如/root/you-get/you-get

初次运行,删除history.db 文件, 修改配置文件config.py

START=1 # 下载系列视频的 第一个
END=1 # 下载系列视频的最后一个 , 比如一个系列教程有30个视频, start=5 ,end = 20 下载从第5个到第20个
ID='BV1oK411L7au' # 视频的ID
YOU_GET_PATH='/home/xda/othergit/you-get/you-get' # 你的you-get路径
MINS=1 # 每次循环等待1分钟
user_id = '518973111' # UP主的ID
total_page = 3 # up主的视频的页数
执行 python downloader.py ,进行下载循环

python people.py ,把某个up主的视频链接加入到待下载队列

python add_data.py --id=BV1oK411L7au --start=4 --end=8 下载视频id为BV1oK411L7au的系列教程,从第4开始,到第8个结束,如果只有一个的话,start和end设为1即可。

可以不断地往队列里面添加下载链接。
主要代码:
# @Time : 2019/1/28 14:19
# @File : youtube_downloader.py

import logging
import os
import subprocess
import datetime
import sqlite3
import time
from config import YOU_GET_PATH,MINS
CMD = 'python {} {}'
filename = 'url.txt'


class SQLite():
def __init__(self):
self.conn = sqlite3.connect('history.db')
self.cursor = self.conn.cursor()
self.create_table()

def create_table(self):
create_sql = 'create table if not exists tb_download (url varchar(100),status tinyint,crawltime datetime)'
create_record_tb = 'create table if not exists tb_record (idx varchar(100) PRIMARY KEY,start tinyint,end tinyint,status tinyint)'
self.cursor.execute(create_record_tb)
self.conn.commit()
self.cursor.execute(create_sql)
self.conn.commit()

def exists(self,url):
querySet = 'select * from tb_download where url = ? and status = 1'
self.cursor.execute(querySet,(url,))
ret = self.cursor.fetchone()
return True if ret else False

def insert_history(self,url,status):
query = 'select * from tb_download where url=?'
self.cursor.execute(query,(url,))
ret = self.cursor.fetchone()
current = datetime.datetime.now()

if ret:
insert_sql='update tb_download set status=?,crawltime=? where url = ?'
args=(status,status,current,url)
else:
insert_sql = 'insert into tb_download values(?,?,?)'
args=(url,status,current)

try:
self.cursor.execute(insert_sql,args)
except:
self.conn.rollback()
return False
else:
self.conn.commit()
return True

def get(self):
sql = 'select idx,start,end from tb_record where status=0'
self.cursor.execute(sql)
ret= self.cursor.fetchone()
return ret

def set(self,idx):
print('set status =1')
sql='update tb_record set status=1 where idx=?'
self.cursor.execute(sql,(idx,))
self.conn.commit()

def llogger(filename):
logger = logging.getLogger(filename) # 不加名称设置root logger

logger.setLevel(logging.DEBUG) # 设置输出级别

formatter = logging.Formatter(
'[%(asctime)s][%(filename)s][line: %(lineno)d]\[%(levelname)s] ## %(message)s)',
datefmt='%Y-%m-%d %H:%M:%S')

# 使用FileHandler输出到文件
prefix = os.path.splitext(filename)[0]
fh = logging.FileHandler(prefix + '.log')
fh.setLevel(logging.DEBUG)
fh.setFormatter(formatter)

# 使用StreamHandler输出到屏幕
ch = logging.StreamHandler()
ch.setLevel(logging.DEBUG)
ch.setFormatter(formatter)

# 添加两个Handler
logger.addHandler(ch)
logger.addHandler(fh)
return logger


logger = llogger('download.log')
sql_obj = SQLite()

def run():
while 1:
result = sql_obj.get()
print(result)
if result:
idx=result[0]
start=result[1]
end=result[2]
try:
download_bilibili(idx,start,end)
except:
pass
else:
sql_obj.set(idx)
else:
time.sleep(MINS*60)

def download_bilibili(id,start_page,total_page):
global doc

bilibili_url = 'https://www.bilibili.com/video/{}?p={}'
for i in range(start_page, total_page+1):

next_url = bilibili_url.format(id, i)
if sql_obj.exists(next_url):
print('have download')
continue

try:
command = CMD.format(YOU_GET_PATH, next_url)
p = subprocess.Popen(command, stdout=subprocess.PIPE, stderr=subprocess.PIPE,
shell=True)

output, error = p.communicate()

except Exception as e:
print('has execption')
sql_obj.insert_history(next_url,status=0)
logger.error(e)
continue
else:
output_str = output.decode()
if len(output_str) == 0:
sql_obj.insert_history(next_url,status=0)
logger.info('下载失败')
continue

logger.info('{} has been downloaded !'.format(next_url))
sql_obj.insert_history(next_url,status=1)

run()

  查看全部
B站上不少优秀的学习资源,下载到本地观看,便于快进,多倍速。 也可以放到平板,手机,在没有网络,或者网络条件不佳的环境下观看。
 

 
使用python实现
https://github.com/Rockyzsu/bilibili
 
B站视频下载
自动批量下载B站一个系列的视频

下载某个UP主的所有视频

使用:

下载you-get库,git clone https://github.com/soimort/you-get.git 复制其本地路径,比如/root/you-get/you-get

初次运行,删除history.db 文件, 修改配置文件config.py

START=1 # 下载系列视频的 第一个
END=1 # 下载系列视频的最后一个 , 比如一个系列教程有30个视频, start=5 ,end = 20 下载从第5个到第20个
ID='BV1oK411L7au' # 视频的ID
YOU_GET_PATH='/home/xda/othergit/you-get/you-get' # 你的you-get路径
MINS=1 # 每次循环等待1分钟
user_id = '518973111' # UP主的ID
total_page = 3 # up主的视频的页数
执行 python downloader.py ,进行下载循环

python people.py ,把某个up主的视频链接加入到待下载队列

python add_data.py --id=BV1oK411L7au --start=4 --end=8 下载视频id为BV1oK411L7au的系列教程,从第4开始,到第8个结束,如果只有一个的话,start和end设为1即可。

可以不断地往队列里面添加下载链接。

主要代码:
# @Time : 2019/1/28 14:19
# @File : youtube_downloader.py

import logging
import os
import subprocess
import datetime
import sqlite3
import time
from config import YOU_GET_PATH,MINS
CMD = 'python {} {}'
filename = 'url.txt'


class SQLite():
def __init__(self):
self.conn = sqlite3.connect('history.db')
self.cursor = self.conn.cursor()
self.create_table()

def create_table(self):
create_sql = 'create table if not exists tb_download (url varchar(100),status tinyint,crawltime datetime)'
create_record_tb = 'create table if not exists tb_record (idx varchar(100) PRIMARY KEY,start tinyint,end tinyint,status tinyint)'
self.cursor.execute(create_record_tb)
self.conn.commit()
self.cursor.execute(create_sql)
self.conn.commit()

def exists(self,url):
querySet = 'select * from tb_download where url = ? and status = 1'
self.cursor.execute(querySet,(url,))
ret = self.cursor.fetchone()
return True if ret else False

def insert_history(self,url,status):
query = 'select * from tb_download where url=?'
self.cursor.execute(query,(url,))
ret = self.cursor.fetchone()
current = datetime.datetime.now()

if ret:
insert_sql='update tb_download set status=?,crawltime=? where url = ?'
args=(status,status,current,url)
else:
insert_sql = 'insert into tb_download values(?,?,?)'
args=(url,status,current)

try:
self.cursor.execute(insert_sql,args)
except:
self.conn.rollback()
return False
else:
self.conn.commit()
return True

def get(self):
sql = 'select idx,start,end from tb_record where status=0'
self.cursor.execute(sql)
ret= self.cursor.fetchone()
return ret

def set(self,idx):
print('set status =1')
sql='update tb_record set status=1 where idx=?'
self.cursor.execute(sql,(idx,))
self.conn.commit()

def llogger(filename):
logger = logging.getLogger(filename) # 不加名称设置root logger

logger.setLevel(logging.DEBUG) # 设置输出级别

formatter = logging.Formatter(
'[%(asctime)s][%(filename)s][line: %(lineno)d]\[%(levelname)s] ## %(message)s)',
datefmt='%Y-%m-%d %H:%M:%S')

# 使用FileHandler输出到文件
prefix = os.path.splitext(filename)[0]
fh = logging.FileHandler(prefix + '.log')
fh.setLevel(logging.DEBUG)
fh.setFormatter(formatter)

# 使用StreamHandler输出到屏幕
ch = logging.StreamHandler()
ch.setLevel(logging.DEBUG)
ch.setFormatter(formatter)

# 添加两个Handler
logger.addHandler(ch)
logger.addHandler(fh)
return logger


logger = llogger('download.log')
sql_obj = SQLite()

def run():
while 1:
result = sql_obj.get()
print(result)
if result:
idx=result[0]
start=result[1]
end=result[2]
try:
download_bilibili(idx,start,end)
except:
pass
else:
sql_obj.set(idx)
else:
time.sleep(MINS*60)

def download_bilibili(id,start_page,total_page):
global doc

bilibili_url = 'https://www.bilibili.com/video/{}?p={}'
for i in range(start_page, total_page+1):

next_url = bilibili_url.format(id, i)
if sql_obj.exists(next_url):
print('have download')
continue

try:
command = CMD.format(YOU_GET_PATH, next_url)
p = subprocess.Popen(command, stdout=subprocess.PIPE, stderr=subprocess.PIPE,
shell=True)

output, error = p.communicate()

except Exception as e:
print('has execption')
sql_obj.insert_history(next_url,status=0)
logger.error(e)
continue
else:
output_str = output.decode()
if len(output_str) == 0:
sql_obj.insert_history(next_url,status=0)
logger.info('下载失败')
continue

logger.info('{} has been downloaded !'.format(next_url))
sql_obj.insert_history(next_url,status=1)

run()

 

开通Ptrade Python量化接口 国金证券/国盛证券

股票绫波丽 发表了文章 • 0 个评论 • 6704 次浏览 • 2021-07-06 08:40 • 来自相关话题

目前不少在A股做量化的大部分使用tushare,优矿,米筐,聚宽等等,无论教程,还是实际操作,基本没见有教怎么用程序下单,实盘交易的。

而退而求其次使用按键精灵,模拟点击交易软件进行点击下单,非常不稳定,无法判断下单后是否成交,也无法实时获取行情数据。如果使用tushare或者新浪接口数据,扫描一次全市场的行情用时很久且不稳定,等扫描结束,再下单,此时价格可能已经是几分钟前的了,且此类接口调用次数多是会被封IP的。

笔者使用的是券商提供的量化软件:Ptrade。是恒生电子研发的提供给机构使用的程序化交易软件。提供策略回测,下单API接口,实时行情获取,并且使用的开发语言python,易于上手。

策略回测与实盘交易




研究页面

研究页面,熟悉python jupyter notebook的朋友对这个界面肯定很熟悉。

研究的页面实际就运行你逐行输出调试程序,了解每个函数的具体使用,或者你策略的中途结果调试。





 
回测策略

实际代码需要在回测策略里面写,写完后确定无误,就可以放在仿真环境下真实运行。如果你运行得到的结果很满意,那么就可以直接部署到实盘服务器上。实盘服务器是在券商那边,不需要个人购买服务器,也不需要本地开着这个Ptrade,就是说不需要在个人电脑上一直开着跑,你的最终代码和程序是在券商服务器上部署与运行,除非有报错异常停止,不然在你不暂停或者停止的前提下,可以一直运行下去。












条件满足后下单

可视化量化

同时也提供一些常见的现成的量化策略,选中后只要鼠标点点点也能够自动化跑这些策略了,当然里面很多参数都可以用鼠标点点点修改。





 
接口文档也非常详细:




 
一些常见策略代码:

集合竞价追涨停策略def initialize(context):
# 初始化此策略
# 设置我们要操作的股票池, 这里我们只操作一支股票
g.security = '600570.SS'
set_universe(g.security)
#每天9:23分运行集合竞价处理函数
run_daily(context, aggregate_auction_func, time='9:23')

def aggregate_auction_func(context):
stock = g.security
#最新价
snapshot = get_snapshot(stock)
price = snapshot[stock]['last_px']
#涨停价
up_limit = snapshot[stock]['up_px']
#如果最新价不小于涨停价,买入
if float(price) >= float(up_limit):
order(g.security, 100, limit_price=up_limit)

def handle_data(context, data):
pass
双均线策略def initialize(context):
# 初始化此策略
# 设置我们要操作的股票池, 这里我们只操作一支股票
g.security = '600570.SS'
set_universe(g.security)
pass

#当五日均线高于十日均线时买入,当五日均线低于十日均线时卖出
def handle_data(context, data):
security = g.security

#得到十日历史价格
df = get_history(10, '1d', 'close', security, fq=None, include=False)

# 得到五日均线价格
ma5 = round(df['close'][-5:].mean(), 3)

# 得到十日均线价格
ma10 = round(df['close'][-10:].mean(), 3)

# 取得昨天收盘价
price = data[security]['close']

# 得到当前资金余额
cash = context.portfolio.cash

# 如果当前有余额,并且五日均线大于十日均线
if ma5 > ma10:
# 用所有 cash 买入股票
order_value(security, cash)
# 记录这次买入
log.info("Buying %s" % (security))

# 如果五日均线小于十日均线,并且目前有头寸
elif ma5 < ma10 and get_position(security).amount > 0:
# 全部卖出
order_target(security, 0)
# 记录这次卖出
log.info("Selling %s" % (security))
 
tick级别均线策略

通俗点就是按照秒级别进行操作。def initialize(context):
# 初始化此策略
# 设置我们要操作的股票池, 这里我们只操作一支股票
g.security = '600570.SS'
set_universe(g.security)
#每3秒运行一次主函数
run_interval(context, func, seconds=3)

#盘前准备历史数据
def before_trading_start(context, data):
history = get_history(10, '1d', 'close', g.security, fq='pre', include=False)
g.close_array = history['close'].values

#当五日均线高于十日均线时买入,当五日均线低于十日均线时卖出
def func(context):

stock = g.security

#获取最新价
snapshot = get_snapshot(stock)
price = snapshot[stock]['last_px']

# 得到五日均线价格
days = 5
ma5 = get_MA_day(stock, days, g.close_array[-4:], price)
# 得到十日均线价格
days = 10
ma10 = get_MA_day(stock, days, g.close_array[-9:], price)

# 得到当前资金余额
cash = context.portfolio.cash

# 如果当前有余额,并且五日均线大于十日均线
if ma5 > ma10:
# 用所有 cash 买入股票
order_value(stock, cash)
# 记录这次买入
log.info("Buying %s" % (stock))

# 如果五日均线小于十日均线,并且目前有头寸
elif ma5 < ma10 and get_position(stock).amount > 0:
# 全部卖出
order_target(stock, 0)
# 记录这次卖出
log.info("Selling %s" % (stock))

#计算实时均线函数
def get_MA_day(stock,days,close_array,current_price):
close_sum = close_array[-(days-1):].sum()
MA = (current_price + close_sum)/days
return MA

def handle_data(context, data):
pass
 
macd策略def f_expma(N,m,EXPMA1,price):
a = m/(N+1)
EXPMA2 = a * price + (1 - a)*EXPMA1
return EXPMA2 #2为后一天值

#定义macd函数,输入平滑系数参数、前一日值,输出当日值
def macd(N1,N2,N3,m,EXPMA12_1,EXPMA26_1,DEA1,price):
EXPMA12_2 = f_expma(N1,m,EXPMA12_1,price)
EXPMA26_2 = f_expma(N2,m,EXPMA26_1,price)
DIF2 = EXPMA12_2 - EXPMA26_2
a = m/(N3+1)
DEA2 = a * DIF2 + (1 - a)*DEA1
BAR2=2*(DIF2-DEA2)
return EXPMA12_2,EXPMA26_2,DIF2,DEA2,BAR2

def initialize(context):
global init_price
init_price = None
# 获取沪深300股票
g.security = get_index_stocks('000300.SS')
#g.security = ['600570.SS']
# 设置我们要操作的股票池, 这里我们只操作一支股票
set_universe(g.security)

def handle_data(context, data):
# 获取历史数据,这里只获取了2天的数据,如果希望最终MACD指标结果更准确最好是获取
# 从股票上市至今的所有历史数据,即增加获取的天数
close_price = get_history(2, '1d', field='close', security_list=g.security)
#如果是停牌不进行计算
for security in g.security:
if data[security].is_open >0:
global init_price,EXPMA12_1,EXPMA26_1,EXPMA12_2,EXPMA26_2,DIF1,DIF2,DEA1,DEA2
if init_price is None:
init_price = close_price[security].mean()#nan和N-1个数,mean为N-1个数的均值
EXPMA12_1 = init_price
EXPMA26_1 = init_price
DIF1 = init_price
DEA1 = init_price
# m用于计算平滑系数a=m/(N+1)
m = 2.0
#设定指数平滑基期数
N1 = 12
N2 = 26
N3 = 9
EXPMA12_2,EXPMA26_2,DIF2,DEA2,BAR2 = macd(N1,N2,N3,m,EXPMA12_1,EXPMA26_1,DEA1,close_price[security][-1])
# 取得当前价格
current_price = data[security].price
# 取得当前的现金
cash = context.portfolio.cash
# DIF、DEA均为正,DIF向上突破DEA,买入信号参考
if DIF2 > 0 and DEA2 > 0 and DIF1 < DEA1 and DIF2 > DEA2:
# 计算可以买多少只股票
number_of_shares = int(cash/current_price)
# 购买量大于0时,下单
if number_of_shares > 0:
# 以市单价买入股票,日回测时即是开盘价
order(security, +number_of_shares)
# 记录这次买入
log.info("Buying %s" % (security))
# DIF、DEA均为负,DIF向下突破DEA,卖出信号参考
elif DIF2 < 0 and DEA2 < 0 and DIF1 > DEA1 and DIF2 < DEA2 and get_position(security).amount > 0:
# 卖出所有股票,使这只股票的最终持有量为0
order_target(security, 0)
# 记录这次卖出
log.info("Selling %s" % (security))
# 将今日的值赋给全局变量作为下一次前一日的值
DEA1 = DEA2
DIF1 = DIF2
EXPMA12_1 = EXPMA12_2
EXPMA26_1 = EXPMA26_2
 
软件与交易接口开通条件:

开通该券商后,存入资金指定资金即可开通。开通后股票交易费率万一

本身券商的交易费率为股票万一,可转债沪百万分之五,深十万分之五,基金万0.5,非常厚道。
 
不太了解量化行业的可以了解下,不少面向机构的量化交易软件的佣金是万2.5的,且开户门槛高,基本是500W以上,比如华泰的matic量化的门槛是1千万元起步。

所以笔者还是很推荐目前该券商的量化交易接口。

需要开通咨询了解的朋友可以扫码联系:


开通券商账户后可以 可以先试用,再考虑是否开通量化接口权限 查看全部
目前不少在A股做量化的大部分使用tushare,优矿,米筐,聚宽等等,无论教程,还是实际操作,基本没见有教怎么用程序下单,实盘交易的。

而退而求其次使用按键精灵,模拟点击交易软件进行点击下单,非常不稳定,无法判断下单后是否成交,也无法实时获取行情数据。如果使用tushare或者新浪接口数据,扫描一次全市场的行情用时很久且不稳定,等扫描结束,再下单,此时价格可能已经是几分钟前的了,且此类接口调用次数多是会被封IP的。

笔者使用的是券商提供的量化软件:Ptrade。是恒生电子研发的提供给机构使用的程序化交易软件。提供策略回测,下单API接口,实时行情获取,并且使用的开发语言python,易于上手。

策略回测与实盘交易
UrD5TUnxZD.png

研究页面

研究页面,熟悉python jupyter notebook的朋友对这个界面肯定很熟悉。

研究的页面实际就运行你逐行输出调试程序,了解每个函数的具体使用,或者你策略的中途结果调试。

Hc8f4UtMfW.png

 
回测策略

实际代码需要在回测策略里面写,写完后确定无误,就可以放在仿真环境下真实运行。如果你运行得到的结果很满意,那么就可以直接部署到实盘服务器上。实盘服务器是在券商那边,不需要个人购买服务器,也不需要本地开着这个Ptrade,就是说不需要在个人电脑上一直开着跑,你的最终代码和程序是在券商服务器上部署与运行,除非有报错异常停止,不然在你不暂停或者停止的前提下,可以一直运行下去。

Pig1iRRQnP.png



25YjBEdOqa.png


条件满足后下单

可视化量化

同时也提供一些常见的现成的量化策略,选中后只要鼠标点点点也能够自动化跑这些策略了,当然里面很多参数都可以用鼠标点点点修改。

kTmn9iOXaS.png

 
接口文档也非常详细:
v4QFDpHNpd.png

 
一些常见策略代码:

集合竞价追涨停策略
def initialize(context):
# 初始化此策略
# 设置我们要操作的股票池, 这里我们只操作一支股票
g.security = '600570.SS'
set_universe(g.security)
#每天9:23分运行集合竞价处理函数
run_daily(context, aggregate_auction_func, time='9:23')

def aggregate_auction_func(context):
stock = g.security
#最新价
snapshot = get_snapshot(stock)
price = snapshot[stock]['last_px']
#涨停价
up_limit = snapshot[stock]['up_px']
#如果最新价不小于涨停价,买入
if float(price) >= float(up_limit):
order(g.security, 100, limit_price=up_limit)

def handle_data(context, data):
pass

双均线策略
def initialize(context):
# 初始化此策略
# 设置我们要操作的股票池, 这里我们只操作一支股票
g.security = '600570.SS'
set_universe(g.security)
pass

#当五日均线高于十日均线时买入,当五日均线低于十日均线时卖出
def handle_data(context, data):
security = g.security

#得到十日历史价格
df = get_history(10, '1d', 'close', security, fq=None, include=False)

# 得到五日均线价格
ma5 = round(df['close'][-5:].mean(), 3)

# 得到十日均线价格
ma10 = round(df['close'][-10:].mean(), 3)

# 取得昨天收盘价
price = data[security]['close']

# 得到当前资金余额
cash = context.portfolio.cash

# 如果当前有余额,并且五日均线大于十日均线
if ma5 > ma10:
# 用所有 cash 买入股票
order_value(security, cash)
# 记录这次买入
log.info("Buying %s" % (security))

# 如果五日均线小于十日均线,并且目前有头寸
elif ma5 < ma10 and get_position(security).amount > 0:
# 全部卖出
order_target(security, 0)
# 记录这次卖出
log.info("Selling %s" % (security))

 
tick级别均线策略

通俗点就是按照秒级别进行操作。
def initialize(context):
# 初始化此策略
# 设置我们要操作的股票池, 这里我们只操作一支股票
g.security = '600570.SS'
set_universe(g.security)
#每3秒运行一次主函数
run_interval(context, func, seconds=3)

#盘前准备历史数据
def before_trading_start(context, data):
history = get_history(10, '1d', 'close', g.security, fq='pre', include=False)
g.close_array = history['close'].values

#当五日均线高于十日均线时买入,当五日均线低于十日均线时卖出
def func(context):

stock = g.security

#获取最新价
snapshot = get_snapshot(stock)
price = snapshot[stock]['last_px']

# 得到五日均线价格
days = 5
ma5 = get_MA_day(stock, days, g.close_array[-4:], price)
# 得到十日均线价格
days = 10
ma10 = get_MA_day(stock, days, g.close_array[-9:], price)

# 得到当前资金余额
cash = context.portfolio.cash

# 如果当前有余额,并且五日均线大于十日均线
if ma5 > ma10:
# 用所有 cash 买入股票
order_value(stock, cash)
# 记录这次买入
log.info("Buying %s" % (stock))

# 如果五日均线小于十日均线,并且目前有头寸
elif ma5 < ma10 and get_position(stock).amount > 0:
# 全部卖出
order_target(stock, 0)
# 记录这次卖出
log.info("Selling %s" % (stock))

#计算实时均线函数
def get_MA_day(stock,days,close_array,current_price):
close_sum = close_array[-(days-1):].sum()
MA = (current_price + close_sum)/days
return MA

def handle_data(context, data):
pass

 
macd策略
def f_expma(N,m,EXPMA1,price):
a = m/(N+1)
EXPMA2 = a * price + (1 - a)*EXPMA1
return EXPMA2 #2为后一天值

#定义macd函数,输入平滑系数参数、前一日值,输出当日值
def macd(N1,N2,N3,m,EXPMA12_1,EXPMA26_1,DEA1,price):
EXPMA12_2 = f_expma(N1,m,EXPMA12_1,price)
EXPMA26_2 = f_expma(N2,m,EXPMA26_1,price)
DIF2 = EXPMA12_2 - EXPMA26_2
a = m/(N3+1)
DEA2 = a * DIF2 + (1 - a)*DEA1
BAR2=2*(DIF2-DEA2)
return EXPMA12_2,EXPMA26_2,DIF2,DEA2,BAR2

def initialize(context):
global init_price
init_price = None
# 获取沪深300股票
g.security = get_index_stocks('000300.SS')
#g.security = ['600570.SS']
# 设置我们要操作的股票池, 这里我们只操作一支股票
set_universe(g.security)

def handle_data(context, data):
# 获取历史数据,这里只获取了2天的数据,如果希望最终MACD指标结果更准确最好是获取
# 从股票上市至今的所有历史数据,即增加获取的天数
close_price = get_history(2, '1d', field='close', security_list=g.security)
#如果是停牌不进行计算
for security in g.security:
if data[security].is_open >0:
global init_price,EXPMA12_1,EXPMA26_1,EXPMA12_2,EXPMA26_2,DIF1,DIF2,DEA1,DEA2
if init_price is None:
init_price = close_price[security].mean()#nan和N-1个数,mean为N-1个数的均值
EXPMA12_1 = init_price
EXPMA26_1 = init_price
DIF1 = init_price
DEA1 = init_price
# m用于计算平滑系数a=m/(N+1)
m = 2.0
#设定指数平滑基期数
N1 = 12
N2 = 26
N3 = 9
EXPMA12_2,EXPMA26_2,DIF2,DEA2,BAR2 = macd(N1,N2,N3,m,EXPMA12_1,EXPMA26_1,DEA1,close_price[security][-1])
# 取得当前价格
current_price = data[security].price
# 取得当前的现金
cash = context.portfolio.cash
# DIF、DEA均为正,DIF向上突破DEA,买入信号参考
if DIF2 > 0 and DEA2 > 0 and DIF1 < DEA1 and DIF2 > DEA2:
# 计算可以买多少只股票
number_of_shares = int(cash/current_price)
# 购买量大于0时,下单
if number_of_shares > 0:
# 以市单价买入股票,日回测时即是开盘价
order(security, +number_of_shares)
# 记录这次买入
log.info("Buying %s" % (security))
# DIF、DEA均为负,DIF向下突破DEA,卖出信号参考
elif DIF2 < 0 and DEA2 < 0 and DIF1 > DEA1 and DIF2 < DEA2 and get_position(security).amount > 0:
# 卖出所有股票,使这只股票的最终持有量为0
order_target(security, 0)
# 记录这次卖出
log.info("Selling %s" % (security))
# 将今日的值赋给全局变量作为下一次前一日的值
DEA1 = DEA2
DIF1 = DIF2
EXPMA12_1 = EXPMA12_2
EXPMA26_1 = EXPMA26_2

 
软件与交易接口开通条件:

开通该券商后,存入资金指定资金即可开通。开通后股票交易费率万一

本身券商的交易费率为股票万一,可转债沪百万分之五,深十万分之五,基金万0.5,非常厚道。
 
不太了解量化行业的可以了解下,不少面向机构的量化交易软件的佣金是万2.5的,且开户门槛高,基本是500W以上,比如华泰的matic量化的门槛是1千万元起步。

所以笔者还是很推荐目前该券商的量化交易接口。

需要开通咨询了解的朋友可以扫码联系:


开通券商账户后可以 可以先试用,再考虑是否开通量化接口权限

量化交易接口python A股

股票绫波丽 发表了文章 • 0 个评论 • 2593 次浏览 • 2021-07-01 11:22 • 来自相关话题

不少在A股做量化的基本千篇一律tushare,优矿,米筐,聚宽等等,无论教程也好。
 
实际操作也好,几乎没见几个人教你怎么用程序下单,实盘交易。


稍微好点的easytrader使用的是模拟点击交易软件进行点击下单,基本无法判断下单后是否成交,也无法实时获取行情数据。别跟我说用tushare或者新浪这些接口数据,除非你做的是低频交易。
 
试试扫描一次全市场的行情看看用时多少?等得到满足条件了,再下单,此时价格可能已经在涨停板上封住了。


笔者使用的是券商提供的量化软件:Ptrade。
 
是恒生电子研发的提供给券商机构使用的程序化交易软件。提供策略回测,自动下单功能,使用的开发语言python。


策略回测与实盘交易






研究页面


研究页面,熟悉python jupyter notebook的朋友对这个界面肯定很熟悉。

研究的页面实际就运行你逐行输出调试程序,了解每个函数的具体使用,或者你策略的中途结果调试。








回测策略

实际代码需要在回测策略里面写,写完后确定无误,就可以放在仿真环境下真实运行。
 
如果你运行得到的结果很满意,那么就可以直接部署到实盘服务器上。
 
实盘服务器是在券商那边,不需要个人购买服务器,也不需要本地开着这个Ptrade,就是说不需要在个人电脑上一直开着跑,你的最终代码和程序是在券商服务器上部署与运行,除非有报错异常停止,不然在你不暂停或者停止的前提下,可以一直运行下去。












条件满足后下单

可视化量化

同时也提供一些常见的现成的量化策略,选中后只要鼠标点点点也能够自动化跑这些策略了,当然里面很多参数都可以用鼠标点点点修改。






接口文档也非常详细:








一些常见策略代码:

双均线策略 def initialize(context):
2 # 初始化此策略
3 # 设置我们要操作的股票池, 这里我们只操作一支股票
4 g.security = '600570.SS'
5 set_universe(g.security)
6 pass
7
8 #当五日均线高于十日均线时买入,当五日均线低于十日均线时卖出
9 def handle_data(context, data):
10 security = g.security
11
12 #得到十日历史价格
13 df = get_history(10, '1d', 'close', security, fq=None, include=False)
14
15 # 得到五日均线价格
16 ma5 = round(df['close'][-5:].mean(), 3)
17
18 # 得到十日均线价格
19 ma10 = round(df['close'][-10:].mean(), 3)
20
21 # 取得昨天收盘价
22 price = data[security]['close']
23
24 # 得到当前资金余额
25 cash = context.portfolio.cash
26
27 # 如果当前有余额,并且五日均线大于十日均线
28 if ma5 > ma10:
29 # 用所有 cash 买入股票
30 order_value(security, cash)
31 # 记录这次买入
32 log.info("Buying %s" % (security))
33
34 # 如果五日均线小于十日均线,并且目前有头寸
35 elif ma5 < ma10 and get_position(security).amount > 0:
36 # 全部卖出
37 order_target(security, 0)
38 # 记录这次卖出
39 log.info("Selling %s" % (security))
 macd策略
指数平滑均线函数,以price计算,可以选择收盘、开盘价等价格,N为时间周期,m用于计算平滑系数a=m/(N+1),EXPMA1为前一日值 def f_expma(N,m,EXPMA1,price):
2 a = m/(N+1)
3 EXPMA2 = a * price + (1 - a)*EXPMA1
4 return EXPMA2 #2为后一天值
5
6 #定义macd函数,输入平滑系数参数、前一日值,输出当日值
7 def macd(N1,N2,N3,m,EXPMA12_1,EXPMA26_1,DEA1,price):
8 EXPMA12_2 = f_expma(N1,m,EXPMA12_1,price)
9 EXPMA26_2 = f_expma(N2,m,EXPMA26_1,price)
10 DIF2 = EXPMA12_2 - EXPMA26_2
11 a = m/(N3+1)
12 DEA2 = a * DIF2 + (1 - a)*DEA1
13 BAR2=2*(DIF2-DEA2)
14 return EXPMA12_2,EXPMA26_2,DIF2,DEA2,BAR2
15
16 def initialize(context):
17 global init_price
18 init_price = None
19 # 获取沪深300股票
20 g.security = get_index_stocks('000300.SS')
21 #g.security = ['600570.SS']
22 # 设置我们要操作的股票池, 这里我们只操作一支股票
23 set_universe(g.security)
24
25 def handle_data(context, data):
26 # 获取历史数据,这里只获取了2天的数据,如果希望最终MACD指标结果更准确最好是获取
27 # 从股票上市至今的所有历史数据,即增加获取的天数
28 close_price = get_history(2, '1d', field='close', security_list=g.security)
29 #如果是停牌不进行计算
30 for security in g.security:
31 if data[security].is_open >0:
32 global init_price,EXPMA12_1,EXPMA26_1,EXPMA12_2,EXPMA26_2,DIF1,DIF2,DEA1,DEA2
33 if init_price is None:
34 init_price = close_price[security].mean()#nan和N-1个数,mean为N-1个数的均值
35 EXPMA12_1 = init_price
36 EXPMA26_1 = init_price
37 DIF1 = init_price
38 DEA1 = init_price
39 # m用于计算平滑系数a=m/(N+1)
40 m = 2.0
41 #设定指数平滑基期数
42 N1 = 12
43 N2 = 26
44 N3 = 9
45 EXPMA12_2,EXPMA26_2,DIF2,DEA2,BAR2 = macd(N1,N2,N3,m,EXPMA12_1,EXPMA26_1,DEA1,close_price[security][-1])
46 # 取得当前价格
47 current_price = data[security].price
48 # 取得当前的现金
49 cash = context.portfolio.cash
50 # DIF、DEA均为正,DIF向上突破DEA,买入信号参考
51 if DIF2 > 0 and DEA2 > 0 and DIF1 < DEA1 and DIF2 > DEA2:
52 # 计算可以买多少只股票
53 number_of_shares = int(cash/current_price)
54 # 购买量大于0时,下单
55 if number_of_shares > 0:
56 # 以市单价买入股票,日回测时即是开盘价
57 order(security, +number_of_shares)
58 # 记录这次买入
59 log.info("Buying %s" % (security))
60 # DIF、DEA均为负,DIF向下突破DEA,卖出信号参考
61 elif DIF2 < 0 and DEA2 < 0 and DIF1 > DEA1 and DIF2 < DEA2 and get_position(security).amount > 0:
62 # 卖出所有股票,使这只股票的最终持有量为0
63 order_target(security, 0)
64 # 记录这次卖出
65 log.info("Selling %s" % (security))
66 # 将今日的值赋给全局变量作为下一次前一日的值
67 DEA1 = DEA2
68 DIF1 = DIF2
69 EXPMA12_1 = EXPMA12_2
70 EXPMA26_1 = EXPMA26_2
 
同时也支持可转债T+0交易,笔者平时也主要是进行的可转债的交易。
 
软件与交易接口开通条件,

开通改券商后,存入资金50W放2周即可开通,开通后可取出。

本身券商的交易费率为股票万一免五,可转债百万分之二,非常厚道,因为不少面向机构的量化交易软件的佣金是万2.5的,笔者之前在国金渠道开的就是,坑的一比。所以笔者还是很推荐目前该券商的量化交易接口。
 
需要可以关注公众号获取开通链接: 查看全部
不少在A股做量化的基本千篇一律tushare,优矿,米筐,聚宽等等,无论教程也好。
 
实际操作也好,几乎没见几个人教你怎么用程序下单,实盘交易。


稍微好点的easytrader使用的是模拟点击交易软件进行点击下单,基本无法判断下单后是否成交,也无法实时获取行情数据。别跟我说用tushare或者新浪这些接口数据,除非你做的是低频交易。
 
试试扫描一次全市场的行情看看用时多少?等得到满足条件了,再下单,此时价格可能已经在涨停板上封住了。


笔者使用的是券商提供的量化软件:Ptrade
 
是恒生电子研发的提供给券商机构使用的程序化交易软件。提供策略回测,自动下单功能,使用的开发语言python。


策略回测与实盘交易

UrD5TUnxZD.png


研究页面


研究页面,熟悉python jupyter notebook的朋友对这个界面肯定很熟悉。

研究的页面实际就运行你逐行输出调试程序,了解每个函数的具体使用,或者你策略的中途结果调试。


Hc8f4UtMfW.png



回测策略

实际代码需要在回测策略里面写,写完后确定无误,就可以放在仿真环境下真实运行。
 
如果你运行得到的结果很满意,那么就可以直接部署到实盘服务器上。
 
实盘服务器是在券商那边,不需要个人购买服务器,也不需要本地开着这个Ptrade,就是说不需要在个人电脑上一直开着跑,你的最终代码和程序是在券商服务器上部署与运行,除非有报错异常停止,不然在你不暂停或者停止的前提下,可以一直运行下去。

Pig1iRRQnP.png



25YjBEdOqa.png


条件满足后下单

可视化量化

同时也提供一些常见的现成的量化策略,选中后只要鼠标点点点也能够自动化跑这些策略了,当然里面很多参数都可以用鼠标点点点修改。

kTmn9iOXaS.png


接口文档也非常详细:


v4QFDpHNpd.png



一些常见策略代码:

双均线策略
    def initialize(context):
2 # 初始化此策略
3 # 设置我们要操作的股票池, 这里我们只操作一支股票
4 g.security = '600570.SS'
5 set_universe(g.security)
6 pass
7
8 #当五日均线高于十日均线时买入,当五日均线低于十日均线时卖出
9 def handle_data(context, data):
10 security = g.security
11
12 #得到十日历史价格
13 df = get_history(10, '1d', 'close', security, fq=None, include=False)
14
15 # 得到五日均线价格
16 ma5 = round(df['close'][-5:].mean(), 3)
17
18 # 得到十日均线价格
19 ma10 = round(df['close'][-10:].mean(), 3)
20
21 # 取得昨天收盘价
22 price = data[security]['close']
23
24 # 得到当前资金余额
25 cash = context.portfolio.cash
26
27 # 如果当前有余额,并且五日均线大于十日均线
28 if ma5 > ma10:
29 # 用所有 cash 买入股票
30 order_value(security, cash)
31 # 记录这次买入
32 log.info("Buying %s" % (security))
33
34 # 如果五日均线小于十日均线,并且目前有头寸
35 elif ma5 < ma10 and get_position(security).amount > 0:
36 # 全部卖出
37 order_target(security, 0)
38 # 记录这次卖出
39 log.info("Selling %s" % (security))

 macd策略
指数平滑均线函数,以price计算,可以选择收盘、开盘价等价格,N为时间周期,m用于计算平滑系数a=m/(N+1),EXPMA1为前一日值
    def f_expma(N,m,EXPMA1,price):
2 a = m/(N+1)
3 EXPMA2 = a * price + (1 - a)*EXPMA1
4 return EXPMA2 #2为后一天值
5
6 #定义macd函数,输入平滑系数参数、前一日值,输出当日值
7 def macd(N1,N2,N3,m,EXPMA12_1,EXPMA26_1,DEA1,price):
8 EXPMA12_2 = f_expma(N1,m,EXPMA12_1,price)
9 EXPMA26_2 = f_expma(N2,m,EXPMA26_1,price)
10 DIF2 = EXPMA12_2 - EXPMA26_2
11 a = m/(N3+1)
12 DEA2 = a * DIF2 + (1 - a)*DEA1
13 BAR2=2*(DIF2-DEA2)
14 return EXPMA12_2,EXPMA26_2,DIF2,DEA2,BAR2
15
16 def initialize(context):
17 global init_price
18 init_price = None
19 # 获取沪深300股票
20 g.security = get_index_stocks('000300.SS')
21 #g.security = ['600570.SS']
22 # 设置我们要操作的股票池, 这里我们只操作一支股票
23 set_universe(g.security)
24
25 def handle_data(context, data):
26 # 获取历史数据,这里只获取了2天的数据,如果希望最终MACD指标结果更准确最好是获取
27 # 从股票上市至今的所有历史数据,即增加获取的天数
28 close_price = get_history(2, '1d', field='close', security_list=g.security)
29 #如果是停牌不进行计算
30 for security in g.security:
31 if data[security].is_open >0:
32 global init_price,EXPMA12_1,EXPMA26_1,EXPMA12_2,EXPMA26_2,DIF1,DIF2,DEA1,DEA2
33 if init_price is None:
34 init_price = close_price[security].mean()#nan和N-1个数,mean为N-1个数的均值
35 EXPMA12_1 = init_price
36 EXPMA26_1 = init_price
37 DIF1 = init_price
38 DEA1 = init_price
39 # m用于计算平滑系数a=m/(N+1)
40 m = 2.0
41 #设定指数平滑基期数
42 N1 = 12
43 N2 = 26
44 N3 = 9
45 EXPMA12_2,EXPMA26_2,DIF2,DEA2,BAR2 = macd(N1,N2,N3,m,EXPMA12_1,EXPMA26_1,DEA1,close_price[security][-1])
46 # 取得当前价格
47 current_price = data[security].price
48 # 取得当前的现金
49 cash = context.portfolio.cash
50 # DIF、DEA均为正,DIF向上突破DEA,买入信号参考
51 if DIF2 > 0 and DEA2 > 0 and DIF1 < DEA1 and DIF2 > DEA2:
52 # 计算可以买多少只股票
53 number_of_shares = int(cash/current_price)
54 # 购买量大于0时,下单
55 if number_of_shares > 0:
56 # 以市单价买入股票,日回测时即是开盘价
57 order(security, +number_of_shares)
58 # 记录这次买入
59 log.info("Buying %s" % (security))
60 # DIF、DEA均为负,DIF向下突破DEA,卖出信号参考
61 elif DIF2 < 0 and DEA2 < 0 and DIF1 > DEA1 and DIF2 < DEA2 and get_position(security).amount > 0:
62 # 卖出所有股票,使这只股票的最终持有量为0
63 order_target(security, 0)
64 # 记录这次卖出
65 log.info("Selling %s" % (security))
66 # 将今日的值赋给全局变量作为下一次前一日的值
67 DEA1 = DEA2
68 DIF1 = DIF2
69 EXPMA12_1 = EXPMA12_2
70 EXPMA26_1 = EXPMA26_2

 
同时也支持可转债T+0交易,笔者平时也主要是进行的可转债的交易。
 
软件与交易接口开通条件,

开通改券商后,存入资金50W放2周即可开通,开通后可取出。

本身券商的交易费率为股票万一免五,可转债百万分之二,非常厚道,因为不少面向机构的量化交易软件的佣金是万2.5的,笔者之前在国金渠道开的就是,坑的一比。所以笔者还是很推荐目前该券商的量化交易接口。
 
需要可以关注公众号获取开通链接:

scrapy源码分析<一>:入口函数以及是如何运行

python爬虫李魔佛 发表了文章 • 0 个评论 • 3876 次浏览 • 2019-08-31 10:47 • 来自相关话题

运行scrapy crawl example 命令的时候,就会执行我们写的爬虫程序。
下面我们从源码分析一下scrapy执行的流程:
 

执行scrapy crawl 命令时,调用的是Command类class Command(ScrapyCommand):

requires_project = True

def syntax(self):
return '[options]'

def short_desc(self):
return 'Runs all of the spiders - My Defined'

def run(self,args,opts):
print('==================')
print(type(self.crawler_process))
spider_list = self.crawler_process.spiders.list() # 找到爬虫类

for name in spider_list:
print('=================')
print(name)
self.crawler_process.crawl(name,**opts.__dict__)

self.crawler_process.start()
然后我们去看看crawler_process,这个是来自ScrapyCommand,而ScrapyCommand又是CrawlerProcess的子类,而CrawlerProcess又是CrawlerRunner的子类

在CrawlerRunner构造函数里面主要作用就是这个 def __init__(self, settings=None):
if isinstance(settings, dict) or settings is None:
settings = Settings(settings)
self.settings = settings
self.spider_loader = _get_spider_loader(settings) # 构造爬虫
self._crawlers = set()
self._active = set()
self.bootstrap_failed = False
1. 加载配置文件def _get_spider_loader(settings):

cls_path = settings.get('SPIDER_LOADER_CLASS')

# settings文件没有定义SPIDER_LOADER_CLASS,所以这里获取到的是系统的默认配置文件,
# 默认配置文件在接下来的代码块A
# SPIDER_LOADER_CLASS = 'scrapy.spiderloader.SpiderLoader'

loader_cls = load_object(cls_path)
# 这个函数就是根据路径转为类对象,也就是上面crapy.spiderloader.SpiderLoader 这个
# 字符串变成一个类对象
# 具体的load_object 对象代码见下面代码块B

return loader_cls.from_settings(settings.frozencopy())
默认配置文件defautl_settting.py# 代码块A
#......省略若干
SCHEDULER = 'scrapy.core.scheduler.Scheduler'
SCHEDULER_DISK_QUEUE = 'scrapy.squeues.PickleLifoDiskQueue'
SCHEDULER_MEMORY_QUEUE = 'scrapy.squeues.LifoMemoryQueue'
SCHEDULER_PRIORITY_QUEUE = 'scrapy.pqueues.ScrapyPriorityQueue'

SPIDER_LOADER_CLASS = 'scrapy.spiderloader.SpiderLoader' 就是这个值
SPIDER_LOADER_WARN_ONLY = False

SPIDER_MIDDLEWARES = {}

load_object的实现# 代码块B 为了方便,我把异常处理的去除
from importlib import import_module #导入第三方库

def load_object(path):
dot = path.rindex('.')
module, name = path[:dot], path[dot+1:]
# 上面把路径分为基本路径+模块名

mod = import_module(module)
obj = getattr(mod, name)
# 获取模块里面那个值

return obj

测试代码:In [33]: mod = import_module(module)

In [34]: mod
Out[34]: <module 'scrapy.spiderloader' from '/home/xda/anaconda3/lib/python3.7/site-packages/scrapy/spiderloader.py'>

In [35]: getattr(mod,name)
Out[35]: scrapy.spiderloader.SpiderLoader

In [36]: obj = getattr(mod,name)

In [37]: obj
Out[37]: scrapy.spiderloader.SpiderLoader

In [38]: type(obj)
Out[38]: type
在代码块A中,loader_cls是SpiderLoader,最后返回的的是SpiderLoader.from_settings(settings.frozencopy())
接下来看看SpiderLoader.from_settings, def from_settings(cls, settings):
return cls(settings)
返回类对象自己,所以直接看__init__函数即可class SpiderLoader(object):
"""
SpiderLoader is a class which locates and loads spiders
in a Scrapy project.
"""
def __init__(self, settings):
self.spider_modules = settings.getlist('SPIDER_MODULES')
# 获得settting中的模块名字,创建scrapy的时候就默认帮你生成了
# 你可以看看你的settings文件里面的内容就可以找到这个值,是一个list

self.warn_only = settings.getbool('SPIDER_LOADER_WARN_ONLY')
self._spiders = {}
self._found = defaultdict(list)
self._load_all_spiders() # 加载所有爬虫

核心就是这个_load_all_spiders:
走起:def _load_all_spiders(self):
for name in self.spider_modules:

for module in walk_modules(name): # 这个遍历文件夹里面的文件,然后再转化为类对象,
# 保存到字典:self._spiders = {}
self._load_spiders(module) # 模块变成spider

self._check_name_duplicates() # 去重,如果名字一样就异常

接下来看看_load_spiders
核心就是下面的。def iter_spider_classes(module):
from scrapy.spiders import Spider

for obj in six.itervalues(vars(module)): # 找到模块里面的变量,然后迭代出来
if inspect.isclass(obj) and \
issubclass(obj, Spider) and \
obj.__module__ == module.__name__ and \
getattr(obj, 'name', None): # 有name属性,继承于Spider
yield obj
这个obj就是我们平时写的spider类了。
原来分析了这么多,才找到了我们平时写的爬虫类

待续。。。。
 
原创文章
转载请注明出处
http://30daydo.com/article/530
  查看全部
运行scrapy crawl example 命令的时候,就会执行我们写的爬虫程序。
下面我们从源码分析一下scrapy执行的流程:
 

执行scrapy crawl 命令时,调用的是Command类
class Command(ScrapyCommand):

requires_project = True

def syntax(self):
return '[options]'

def short_desc(self):
return 'Runs all of the spiders - My Defined'

def run(self,args,opts):
print('==================')
print(type(self.crawler_process))
spider_list = self.crawler_process.spiders.list() # 找到爬虫类

for name in spider_list:
print('=================')
print(name)
self.crawler_process.crawl(name,**opts.__dict__)

self.crawler_process.start()

然后我们去看看crawler_process,这个是来自ScrapyCommand,而ScrapyCommand又是CrawlerProcess的子类,而CrawlerProcess又是CrawlerRunner的子类

在CrawlerRunner构造函数里面主要作用就是这个
      def __init__(self, settings=None):
if isinstance(settings, dict) or settings is None:
settings = Settings(settings)
self.settings = settings
self.spider_loader = _get_spider_loader(settings) # 构造爬虫
self._crawlers = set()
self._active = set()
self.bootstrap_failed = False

1. 加载配置文件
def _get_spider_loader(settings):

cls_path = settings.get('SPIDER_LOADER_CLASS')

# settings文件没有定义SPIDER_LOADER_CLASS,所以这里获取到的是系统的默认配置文件,
# 默认配置文件在接下来的代码块A
# SPIDER_LOADER_CLASS = 'scrapy.spiderloader.SpiderLoader'

loader_cls = load_object(cls_path)
# 这个函数就是根据路径转为类对象,也就是上面crapy.spiderloader.SpiderLoader 这个
# 字符串变成一个类对象
# 具体的load_object 对象代码见下面代码块B

return loader_cls.from_settings(settings.frozencopy())

默认配置文件defautl_settting.py
# 代码块A
#......省略若干
SCHEDULER = 'scrapy.core.scheduler.Scheduler'
SCHEDULER_DISK_QUEUE = 'scrapy.squeues.PickleLifoDiskQueue'
SCHEDULER_MEMORY_QUEUE = 'scrapy.squeues.LifoMemoryQueue'
SCHEDULER_PRIORITY_QUEUE = 'scrapy.pqueues.ScrapyPriorityQueue'

SPIDER_LOADER_CLASS = 'scrapy.spiderloader.SpiderLoader' 就是这个值
SPIDER_LOADER_WARN_ONLY = False

SPIDER_MIDDLEWARES = {}


load_object的实现
# 代码块B 为了方便,我把异常处理的去除
from importlib import import_module #导入第三方库

def load_object(path):
dot = path.rindex('.')
module, name = path[:dot], path[dot+1:]
# 上面把路径分为基本路径+模块名

mod = import_module(module)
obj = getattr(mod, name)
# 获取模块里面那个值

return obj


测试代码:
In [33]: mod = import_module(module)                                                                                                                                             

In [34]: mod
Out[34]: <module 'scrapy.spiderloader' from '/home/xda/anaconda3/lib/python3.7/site-packages/scrapy/spiderloader.py'>

In [35]: getattr(mod,name)
Out[35]: scrapy.spiderloader.SpiderLoader

In [36]: obj = getattr(mod,name)

In [37]: obj
Out[37]: scrapy.spiderloader.SpiderLoader

In [38]: type(obj)
Out[38]: type

在代码块A中,loader_cls是SpiderLoader,最后返回的的是SpiderLoader.from_settings(settings.frozencopy())
接下来看看SpiderLoader.from_settings,
    def from_settings(cls, settings):
return cls(settings)

返回类对象自己,所以直接看__init__函数即可
class SpiderLoader(object):
"""
SpiderLoader is a class which locates and loads spiders
in a Scrapy project.
"""
def __init__(self, settings):
self.spider_modules = settings.getlist('SPIDER_MODULES')
# 获得settting中的模块名字,创建scrapy的时候就默认帮你生成了
# 你可以看看你的settings文件里面的内容就可以找到这个值,是一个list

self.warn_only = settings.getbool('SPIDER_LOADER_WARN_ONLY')
self._spiders = {}
self._found = defaultdict(list)
self._load_all_spiders() # 加载所有爬虫


核心就是这个_load_all_spiders:
走起:
def _load_all_spiders(self):
for name in self.spider_modules:

for module in walk_modules(name): # 这个遍历文件夹里面的文件,然后再转化为类对象,
# 保存到字典:self._spiders = {}
self._load_spiders(module) # 模块变成spider

self._check_name_duplicates() # 去重,如果名字一样就异常


接下来看看_load_spiders
核心就是下面的。
def iter_spider_classes(module):
from scrapy.spiders import Spider

for obj in six.itervalues(vars(module)): # 找到模块里面的变量,然后迭代出来
if inspect.isclass(obj) and \
issubclass(obj, Spider) and \
obj.__module__ == module.__name__ and \
getattr(obj, 'name', None): # 有name属性,继承于Spider
yield obj

这个obj就是我们平时写的spider类了。
原来分析了这么多,才找到了我们平时写的爬虫类

待续。。。。
 
原创文章
转载请注明出处
http://30daydo.com/article/530
 

Linux下自制有道词典 - python 解密有道词典JS加密

python爬虫李魔佛 发表了文章 • 0 个评论 • 3294 次浏览 • 2019-02-23 20:17 • 来自相关话题

对于爬虫新手来说,JS解密是一道过不去的坎,需要不断地练习。
平时在linux下开发,鉴于没有什么好用翻译软件,打开网易也占用系统资源,所以写了个在控制台的翻译软件接口。
 
使用python爬虫,查看网页的JS加密方法,一步一步地分析,就能够得到最后的加密方法啦。
 
直接给出代码:
 # -*- coding: utf-8 -*-
# website: http://30daydo.com
# @Time : 2019/2/23 19:34
# @File : youdao.py
# 解密有道词典的JS


import hashlib
import random
import requests
import time


def md5_(word):
s = bytes(word, encoding='utf8')
m = hashlib.md5()
m.update(s)
ret = m.hexdigest()
return ret

def get_sign(word, salt):
ret = md5_('fanyideskweb' + word + salt + 'p09@Bn{h02_BIEe]$P^nG')
return ret


def youdao(word):
url = 'http://fanyi.youdao.com/translate_o?smartresult=dict&smartresult=rule'
headers = {
'Host': 'fanyi.youdao.com',
'User-Agent': 'Mozilla/5.0 (Windows NT 6.1; WOW64; rv:47.0) Gecko/20100101 Firefox/47.0',
'Accept': 'application/json, text/javascript, */*; q=0.01',
'Accept-Language': 'zh-CN,zh;q=0.8,en-US;q=0.5,en;q=0.3',
'Accept-Encoding': 'gzip, deflate',
'Content-Type': 'application/x-www-form-urlencoded; charset=UTF-8',
'X-Requested-With': 'XMLHttpRequest',
'Referer': 'http://fanyi.youdao.com/',
'Content-Length': '252',
'Cookie': 'YOUDAO_MOBILE_ACCESS_TYPE=1; OUTFOX_SEARCH_USER_ID=1672542763@10.169.0.83; JSESSIONID=aaaWzxpjeDu1gbhopLzKw; ___rl__test__cookies=1550913722828; OUTFOX_SEARCH_USER_ID_NCOO=372126049.6326876',
'Connection': 'keep-alive',
'Pragma': 'no-cache',
'Cache-Control': 'no-cache',
}

ts = str(int(time.time()*1000))
salt=ts+str(random.randint(0,10))
bv = md5_("5.0 (Windows)")
sign= get_sign(word,salt)

post_data = {
'i': word,
'from': 'AUTO', 'to': 'AUTO', 'smartresult': 'dict', 'client': 'fanyideskweb', 'salt': salt,
'sign': sign, 'ts': ts, 'bv': bv, 'doctype': 'json', 'version': '2.1',
'keyfrom': 'fanyi.web', 'action': 'FY_BY_REALTIME', 'typoResult': 'false'
}

r = requests.post(
url=url,
headers=headers,
data=post_data
)

for item in r.json().get('smartResult',{}).get('entries'):
print(item)

word='student'
youdao(word)
得到结果:





 
Github:
https://github.com/Rockyzsu/CrawlMan/tree/master/youdao_dictionary
原创文章,转载请注明出处
http://30daydo.com/article/416 查看全部
对于爬虫新手来说,JS解密是一道过不去的坎,需要不断地练习。
平时在linux下开发,鉴于没有什么好用翻译软件,打开网易也占用系统资源,所以写了个在控制台的翻译软件接口。
 
使用python爬虫,查看网页的JS加密方法,一步一步地分析,就能够得到最后的加密方法啦。
 
直接给出代码:
 
# -*- coding: utf-8 -*-
# website: http://30daydo.com
# @Time : 2019/2/23 19:34
# @File : youdao.py
# 解密有道词典的JS


import hashlib
import random
import requests
import time


def md5_(word):
s = bytes(word, encoding='utf8')
m = hashlib.md5()
m.update(s)
ret = m.hexdigest()
return ret

def get_sign(word, salt):
ret = md5_('fanyideskweb' + word + salt + 'p09@Bn{h02_BIEe]$P^nG')
return ret


def youdao(word):
url = 'http://fanyi.youdao.com/translate_o?smartresult=dict&smartresult=rule'
headers = {
'Host': 'fanyi.youdao.com',
'User-Agent': 'Mozilla/5.0 (Windows NT 6.1; WOW64; rv:47.0) Gecko/20100101 Firefox/47.0',
'Accept': 'application/json, text/javascript, */*; q=0.01',
'Accept-Language': 'zh-CN,zh;q=0.8,en-US;q=0.5,en;q=0.3',
'Accept-Encoding': 'gzip, deflate',
'Content-Type': 'application/x-www-form-urlencoded; charset=UTF-8',
'X-Requested-With': 'XMLHttpRequest',
'Referer': 'http://fanyi.youdao.com/',
'Content-Length': '252',
'Cookie': 'YOUDAO_MOBILE_ACCESS_TYPE=1; OUTFOX_SEARCH_USER_ID=1672542763@10.169.0.83; JSESSIONID=aaaWzxpjeDu1gbhopLzKw; ___rl__test__cookies=1550913722828; OUTFOX_SEARCH_USER_ID_NCOO=372126049.6326876',
'Connection': 'keep-alive',
'Pragma': 'no-cache',
'Cache-Control': 'no-cache',
}

ts = str(int(time.time()*1000))
salt=ts+str(random.randint(0,10))
bv = md5_("5.0 (Windows)")
sign= get_sign(word,salt)

post_data = {
'i': word,
'from': 'AUTO', 'to': 'AUTO', 'smartresult': 'dict', 'client': 'fanyideskweb', 'salt': salt,
'sign': sign, 'ts': ts, 'bv': bv, 'doctype': 'json', 'version': '2.1',
'keyfrom': 'fanyi.web', 'action': 'FY_BY_REALTIME', 'typoResult': 'false'
}

r = requests.post(
url=url,
headers=headers,
data=post_data
)

for item in r.json().get('smartResult',{}).get('entries'):
print(item)

word='student'
youdao(word)

得到结果:

youdao.PNG

 
Github:
https://github.com/Rockyzsu/CrawlMan/tree/master/youdao_dictionary
原创文章,转载请注明出处
http://30daydo.com/article/416

python 中文图片文字识别

python李魔佛 发表了文章 • 0 个评论 • 4726 次浏览 • 2019-02-01 10:47 • 来自相关话题

pytesseract这个库识别率偏低,也就菜鸟才会用。
使用方法很简单,安装好pytesseract(里面很多坑,小白的话不可避免要折腾一番),然后下载一个中文的字库,百度网盘:https://pan.baidu.com/s/1_jom2d95IeR40gsvkhUuvQ
 
然后把文件放到tesseract的文件夹中 C:\Program Files (x86)\Tesseract-OCR\tessdata 
然后就可以拿来识别了:
from PIL import Image
im = Image.open('chinese.jpg')
plt.figure(figsize=(20,20))
plt.imshow(im)

pytesseract.image_to_string(im,lang='chi_sim')
图片的内容是这样的:





 
然后识别效果如下:
 
'可L又使用以下的语句i上图片显示大 此'
还是不咋地。
 
那么可以换成大厂的API。试试百度的:
""" 读取图片 """
def get_file_content(filePath):
with open(filePath, 'rb') as fp:
return fp.read()

image = get_file_content('example.jpg')

""" 调用通用文字识别, 图片参数为本地图片 """
client.basicGeneral(image);

""" 如果有可选参数 """
options = {}
options["language_type"] = "CHN_ENG"
options["detect_direction"] = "true"
options["detect_language"] = "true"
options["probability"] = "true"

from aip import AipOcr

""" 你的 APPID AK SK """
APP_ID = '你的 App ID'
API_KEY = '你的 Api Key'
SECRET_KEY = '你的 Secret Key'

client = AipOcr(APP_ID, API_KEY, SECRET_KEY)


""" 带参数调用通用文字识别, 图片参数为本地图片 """
client.basicGeneral(image, options)

url = "https//www.x.com/sample.jpg"

""" 调用通用文字识别, 图片参数为远程url图片 """
client.basicGeneralUrl(url);

""" 如果有可选参数 """
options = {}
options["language_type"] = "CHN_ENG"
options["detect_direction"] = "true"
options["detect_language"] = "true"
options["probability"] = "true"

""" 带参数调用通用文字识别, 图片参数为远程url图片 """
client.basicGeneralUrl(url, options)
先去百度云申请一个API,免费的。
https://cloud.baidu.com/doc/OCR/OCR-Python-SDK.html#.E9.85.8D.E7.BD.AEAipOcr
然后把key复制到上面的代码中就可以了。
 
然后再调用看看结果:
可以使用以下的语句让图片显示大些正确率明显高很多了。
 
 
 
  查看全部
pytesseract这个库识别率偏低,也就菜鸟才会用。
使用方法很简单,安装好pytesseract(里面很多坑,小白的话不可避免要折腾一番),然后下载一个中文的字库,百度网盘:https://pan.baidu.com/s/1_jom2d95IeR40gsvkhUuvQ
 
然后把文件放到tesseract的文件夹中 C:\Program Files (x86)\Tesseract-OCR\tessdata 
然后就可以拿来识别了:
from PIL import Image
im = Image.open('chinese.jpg')
plt.figure(figsize=(20,20))
plt.imshow(im)

pytesseract.image_to_string(im,lang='chi_sim')

图片的内容是这样的:

中文1.JPG

 
然后识别效果如下:
 
'可L又使用以下的语句i上图片显示大 此'

还是不咋地。
 
那么可以换成大厂的API。试试百度的:
""" 读取图片 """
def get_file_content(filePath):
with open(filePath, 'rb') as fp:
return fp.read()

image = get_file_content('example.jpg')

""" 调用通用文字识别, 图片参数为本地图片 """
client.basicGeneral(image);

""" 如果有可选参数 """
options = {}
options["language_type"] = "CHN_ENG"
options["detect_direction"] = "true"
options["detect_language"] = "true"
options["probability"] = "true"

from aip import AipOcr

""" 你的 APPID AK SK """
APP_ID = '你的 App ID'
API_KEY = '你的 Api Key'
SECRET_KEY = '你的 Secret Key'

client = AipOcr(APP_ID, API_KEY, SECRET_KEY)


""" 带参数调用通用文字识别, 图片参数为本地图片 """
client.basicGeneral(image, options)

url = "https//www.x.com/sample.jpg"

""" 调用通用文字识别, 图片参数为远程url图片 """
client.basicGeneralUrl(url);

""" 如果有可选参数 """
options = {}
options["language_type"] = "CHN_ENG"
options["detect_direction"] = "true"
options["detect_language"] = "true"
options["probability"] = "true"

""" 带参数调用通用文字识别, 图片参数为远程url图片 """
client.basicGeneralUrl(url, options)

先去百度云申请一个API,免费的。
https://cloud.baidu.com/doc/OCR/OCR-Python-SDK.html#.E9.85.8D.E7.BD.AEAipOcr
然后把key复制到上面的代码中就可以了。
 
然后再调用看看结果:
可以使用以下的语句让图片显示大些
正确率明显高很多了。
 
 
 
 

可转债价格分布堆叠图 绘制 可视化 python+pyecharts

量化交易李魔佛 发表了文章 • 0 个评论 • 5054 次浏览 • 2019-01-30 10:59 • 来自相关话题

这一节课带大家学习如何利用可视化,更好的呈现数据。
即使你有很多数据,可是,你无法直观地看到数据的总体趋势。使用可视化的绘图,可以帮助我们看到数据背后看不到的数据。 比如我已经有每一个可转债的价格,评级。数据如下:





 点击查看大图

如果我用下面的图形就可以看出规律:




 点击查看大图

横坐标是价格,纵坐标是落在该价格的可转债数量,不同颜色代表不同评级的可转债。
 
可以看到大部分AA-评级(浅橙色)的可转债价格都在100元以下,而AA(浅蓝色)的可转债价格分布较为平均,从90到110都有。而AA+和AAA的一般都在100以上。
 
那么如何使用代码实现呢?from setting import get_mysql_conn,get_engine
import pandas as pd
import pymongo
from pyecharts import Geo,Style,Map
engine = get_engine('db_stock',local='local')
# 堆叠图
from pyecharts import Bar
df = pd.read_sql('tb_bond_jisilu',con=engine)

result ={}
for name,grades in df.groupby('评级'):
# print(name,grades[['可转债名称','可转债价格']])
for each in grades['可转债价格']:
result.setdefault(name,)
result[name].append(each)


# 确定价格的范围

value = [str(i) for i in range(85,140)]
ret = [0]*len(value)
ret1 = dict(zip(value,ret))

ret_A_add = ret1.copy()
for item in result['A+']:
for k in ret1:
if float(k)+0.5>item and float(k)-0.5<=item:
ret_A_add[k]+=1

retAA_ = ret1.copy()
for item in result['']:
for k in ret1:
if float(k)+0.5>item and float(k)-0.5<=item:
retAA_[k]+=1

retAA = ret1.copy()
for item in result['AA']:
for k in ret1:
if float(k)+0.5>item and float(k)-0.5<=item:
retAA[k]+=1

retAA_add = ret1.copy()
for item in result['AA+']:
for k in ret1:
if float(k)+0.5>item and float(k)-0.5<=item:
retAA_add[k]+=1

retAAA = ret1.copy()
for item in result['AAA']:
for k in ret1:
if float(k)+0.5>item and float(k)-0.5<=item:
retAAA[k]+=1

bar = Bar('可转债价格分布')
bar.add('A+',value,list(ret_A_add.values()),is_stack=True,yaxis_max=11)
bar.add('',value,list(retAA_.values()),is_stack=True,yaxis_max=11)
bar.add('AA',value,list(retAA.values()),is_stack=True,yaxis_max=11)
bar.add('AA+',value,list(retAA_add.values()),is_stack=True,yaxis_max=11)
bar.add('AAA',value,list(retAAA.values()),is_stack=True,yaxis_max=11)
如果没有安装pyecharts,需要用pip安装即可。
 

 

原创文章
转载请注明出处:
 http://30daydo.com/article/400 

  查看全部
这一节课带大家学习如何利用可视化,更好的呈现数据。
即使你有很多数据,可是,你无法直观地看到数据的总体趋势。使用可视化的绘图,可以帮助我们看到数据背后看不到的数据。 比如我已经有每一个可转债的价格,评级。数据如下:

可转债数据.JPG

 点击查看大图

如果我用下面的图形就可以看出规律:
可转债价格分布.JPG

 点击查看大图

横坐标是价格,纵坐标是落在该价格的可转债数量,不同颜色代表不同评级的可转债。
 
可以看到大部分AA-评级(浅橙色)的可转债价格都在100元以下,而AA(浅蓝色)的可转债价格分布较为平均,从90到110都有。而AA+和AAA的一般都在100以上。
 
那么如何使用代码实现呢?
from  setting import get_mysql_conn,get_engine
import pandas as pd
import pymongo
from pyecharts import Geo,Style,Map
engine = get_engine('db_stock',local='local')
# 堆叠图
from pyecharts import Bar
df = pd.read_sql('tb_bond_jisilu',con=engine)

result ={}
for name,grades in df.groupby('评级'):
# print(name,grades[['可转债名称','可转债价格']])
for each in grades['可转债价格']:
result.setdefault(name,)
result[name].append(each)


# 确定价格的范围

value = [str(i) for i in range(85,140)]
ret = [0]*len(value)
ret1 = dict(zip(value,ret))

ret_A_add = ret1.copy()
for item in result['A+']:
for k in ret1:
if float(k)+0.5>item and float(k)-0.5<=item:
ret_A_add[k]+=1

retAA_ = ret1.copy()
for item in result['']:
for k in ret1:
if float(k)+0.5>item and float(k)-0.5<=item:
retAA_[k]+=1

retAA = ret1.copy()
for item in result['AA']:
for k in ret1:
if float(k)+0.5>item and float(k)-0.5<=item:
retAA[k]+=1

retAA_add = ret1.copy()
for item in result['AA+']:
for k in ret1:
if float(k)+0.5>item and float(k)-0.5<=item:
retAA_add[k]+=1

retAAA = ret1.copy()
for item in result['AAA']:
for k in ret1:
if float(k)+0.5>item and float(k)-0.5<=item:
retAAA[k]+=1

bar = Bar('可转债价格分布')
bar.add('A+',value,list(ret_A_add.values()),is_stack=True,yaxis_max=11)
bar.add('',value,list(retAA_.values()),is_stack=True,yaxis_max=11)
bar.add('AA',value,list(retAA.values()),is_stack=True,yaxis_max=11)
bar.add('AA+',value,list(retAA_add.values()),is_stack=True,yaxis_max=11)
bar.add('AAA',value,list(retAAA.values()),is_stack=True,yaxis_max=11)

如果没有安装pyecharts,需要用pip安装即可。
 

 

原创文章
转载请注明出处:
 http://30daydo.com/article/400 

 

可转债套利【一】 python找出折价可转债个股

量化交易李魔佛 发表了文章 • 9 个评论 • 17508 次浏览 • 2018-03-16 17:17 • 来自相关话题

关于可转债的定义,可以到https://xueqiu.com/6832369826/103042836 这里科普一下。
 
下面的内容默认你对可转债已经有一定的了解。
 
可转债的价值=正股价格/转股价格 + 利息,忽略可转债的利息,直接用公式 可转债的价值=正股价格/转股价格 计算可转债的价值。
 
如果当前可转债的交易价格(在交易软件上显示的价格)如:




所以万信转债的价格是121.5元,然后万信转债的价值呢? 按照上面的公式,万信转债的正股是万达信息,今天万达信息  (2018-03-16)的股价是





以收盘价为例,17.25。
 
而万信转债的股转价格呢? 这个可以到万信转债F10页面的公告中找到,为13.11元。 所以万信转债的价值是
17.25/13.11 = 1.315 , 可转债单位是100, 所以万信转债的内在价值是1.315*100=131.5, 而当前的交易价格为 121.5





 
 
也就是你用121.5元买到一个价值 131.5的商品, 所以相当于打折买到了一个超值的商品,所以当前的万信转债是折价状态。
 
所以本次任务就是要找出可交易的可转债中折价状态的可转债。
 
然后直接上干货。上python代码。#-*-coding=utf-8
'''
可转债监控
'''
import tushare as ts
from setting import get_engine
engine = get_engine('db_bond')
import pandas as pd
import datetime
class ConvertBond():

def __init__(self):
self.conn=ts.get_apis()
self.allBonds=ts.new_cbonds(pause=2)
self.onSellBond=self.allBonds.dropna(subset=['marketprice'])
self.today=datetime.datetime.now().strftime('%Y-%m-%d %H:%M')

def stockPrice(self,code):
stock_df = ts.get_realtime_quotes(code)
price = float(stock_df['price'].values[0])
return price

def dataframe(self):
price_list=
for code in self.onSellBond['scode']:
price_list.append(self.stockPrice(code))
self.onSellBond['stock_price']=price_list
self.onSellBond['ratio'] = (
self.onSellBond['marketprice']
/(self.onSellBond['stock_price'] / self.onSellBond['convprice'])-1)*100
self.onSellBond['Updated']=self.today
self.onSellBond.to_sql('tb_bond',engine,if_exists='replace')

def closed(self):
ts.close_apis(self.conn)

def main():
bond=ConvertBond()
bond.dataframe()
bond.closed()
if __name__=='__main__':
main()








 上面的setting库,把下面的*** 替换成你自己的Mysql用户和密码即可。import os
import MySQLdb
MYSQL_USER = *********
MYSQL_PASSWORD = ********
MYSQL_HOST = *********
MYSQL_PORT = *****

def get_engine(db):
engine = create_engine('mysql+pymysql://{}:{}@{}:{}/{}?charset=utf8'.format(MYSQL_USER, MYSQL_PASSWORD, MYSQL_HOST, MYSQL_PORT, db))
return engine 
上面的少于100行的代码就能够满足你的要求。
运行后会把结果保存在MySQL 数据库。如下图所示:







点击放大
  2018-03-16 可转债表格
 
其中折价率是ratio列。按照ratio列进行排列,只有2个是正,也就是当前市场是只有2只可转债是处于折价状态的,其余的都是溢价状态(价格比内在价值要贵,忽略利息的前提下,如果把4~5%的利息也算进去的话,-3~4%的折价率其实也算小折价吧)
 
目前万信转债折价10个点,宝信转债折价5.8个点。 所以适合低风险投资者建仓。 因为可转债有兜底价格,所以出现亏损的概率很低(除非遇到黑天鹅,公司破产了,像遇到乐视这种PPT独角兽公司,欠债不还的。 但是A股上能够有资格发行可转债的,本身对公司的盈利,分红都有硬性要求)。
 
所以可以保存上面的代码,可以每天运行一次,可以很方便地找出折价的个股,当然也可以在盘中一直监测,因为可转债的价格是实时变化的,一旦遇到大跌,跌到折价状态,你也可以择时入手标的。

原文链接:
http://30daydo.com/article/286
转载请注明出处
 
可转债低费率,沪市百万分之二,深圳十万分之四,免五 开户
加微信开通

  查看全部
关于可转债的定义,可以到https://xueqiu.com/6832369826/103042836 这里科普一下。
 
下面的内容默认你对可转债已经有一定的了解。
 
可转债的价值=正股价格/转股价格 + 利息,忽略可转债的利息,直接用公式 可转债的价值=正股价格/转股价格 计算可转债的价值。
 
如果当前可转债的交易价格(在交易软件上显示的价格)如:
wxzz.GIF

所以万信转债的价格是121.5元,然后万信转债的价值呢? 按照上面的公式,万信转债的正股是万达信息,今天万达信息  (2018-03-16)的股价是

万达信息.GIF

以收盘价为例,17.25。
 
而万信转债的股转价格呢? 这个可以到万信转债F10页面的公告中找到,为13.11元。 所以万信转债的价值是
17.25/13.11 = 1.315 , 可转债单位是100, 所以万信转债的内在价值是1.315*100=131.5, 而当前的交易价格为 121.5

wxzz.GIF

 
 
也就是你用121.5元买到一个价值 131.5的商品, 所以相当于打折买到了一个超值的商品,所以当前的万信转债是折价状态。
 
所以本次任务就是要找出可交易的可转债中折价状态的可转债。
 
然后直接上干货。上python代码。
#-*-coding=utf-8
'''
可转债监控
'''
import tushare as ts
from setting import get_engine
engine = get_engine('db_bond')
import pandas as pd
import datetime
class ConvertBond():

def __init__(self):
self.conn=ts.get_apis()
self.allBonds=ts.new_cbonds(pause=2)
self.onSellBond=self.allBonds.dropna(subset=['marketprice'])
self.today=datetime.datetime.now().strftime('%Y-%m-%d %H:%M')

def stockPrice(self,code):
stock_df = ts.get_realtime_quotes(code)
price = float(stock_df['price'].values[0])
return price

def dataframe(self):
price_list=
for code in self.onSellBond['scode']:
price_list.append(self.stockPrice(code))
self.onSellBond['stock_price']=price_list
self.onSellBond['ratio'] = (
self.onSellBond['marketprice']
/(self.onSellBond['stock_price'] / self.onSellBond['convprice'])-1)*100
self.onSellBond['Updated']=self.today
self.onSellBond.to_sql('tb_bond',engine,if_exists='replace')

def closed(self):
ts.close_apis(self.conn)

def main():
bond=ConvertBond()
bond.dataframe()
bond.closed()
if __name__=='__main__':
main()








 上面的setting库,把下面的*** 替换成你自己的Mysql用户和密码即可。
import os
import MySQLdb
MYSQL_USER = *********
MYSQL_PASSWORD = ********
MYSQL_HOST = *********
MYSQL_PORT = *****

def get_engine(db):
engine = create_engine('mysql+pymysql://{}:{}@{}:{}/{}?charset=utf8'.format(MYSQL_USER, MYSQL_PASSWORD, MYSQL_HOST, MYSQL_PORT, db))
return engine
 
上面的少于100行的代码就能够满足你的要求。
运行后会把结果保存在MySQL 数据库。如下图所示:


Screenshot_from_2018-03-28_09-14-35.png


点击放大
  2018-03-16 可转债表格
 
其中折价率是ratio列。按照ratio列进行排列,只有2个是正,也就是当前市场是只有2只可转债是处于折价状态的,其余的都是溢价状态(价格比内在价值要贵,忽略利息的前提下,如果把4~5%的利息也算进去的话,-3~4%的折价率其实也算小折价吧)
 
目前万信转债折价10个点,宝信转债折价5.8个点。 所以适合低风险投资者建仓。 因为可转债有兜底价格,所以出现亏损的概率很低(除非遇到黑天鹅,公司破产了,像遇到乐视这种PPT独角兽公司,欠债不还的。 但是A股上能够有资格发行可转债的,本身对公司的盈利,分红都有硬性要求)。
 
所以可以保存上面的代码,可以每天运行一次,可以很方便地找出折价的个股,当然也可以在盘中一直监测,因为可转债的价格是实时变化的,一旦遇到大跌,跌到折价状态,你也可以择时入手标的。

原文链接:
http://30daydo.com/article/286
转载请注明出处
 
可转债低费率,沪市百万分之二,深圳十万分之四,免五 开户
加微信开通

 

dataframe reindex和reset_index区别

量化交易李魔佛 发表了文章 • 0 个评论 • 75041 次浏览 • 2017-12-30 15:58 • 来自相关话题

reset_index的作用是重新设置dataframe的index,范围为0~len(df)。 df = pd.DataFrame({'A': [1, 2, 3, 4, 5], 'B': [10, 20, 30, 40, 50]})
df2 = pd.DataFrame({'A': [6], 'B': [60]})
print 'df\n', df
print 'df2\n', df2

df_x = [df, df2]
result = pd.concat(df_x)
print 'first result\n', result 
上面代码把df和df2合并为一个result,但是result的index是乱的。





 
那么执行result2= result.reset_index()
得到如下的result2: (默认只是返回一个copy,原来的result没有发生改变,所以需要副本赋值给result2)





可以看到,原来的一列index现在变成了columns之一,新的index为[0,1,2,3,4,5]
如果添加参数 reset_index(drop=True) 那么原index会被丢弃,不会显示为一个新列。result2 = result.reset_index(drop=True)



 
reindex的作用是按照原有的列进行重新生成一个新的df。
 
还是使用上面的代码
result目前是df和df2的合并序列。
如下:




 
可以看到index为[0,1,2,3,4,0]
执行 result3 = result.reindex(columns=['A','C'])




 
可以看到,原index并没有发生改变,而列变成了A和C,因为C是不存在的,所以使用了NaB填充,这个值的内容可以自己填充,可以改为默认填充0或者任意你想要的数据。reindex(columns=..)的作用类似于重新把列的顺序整理一遍, 而使用reindex(index=....) 则按照行重新整理一遍。

原文链接:http://30daydo.com/article/257 
欢迎转载,注明出处
  查看全部
reset_index的作用是重新设置dataframe的index,范围为0~len(df)。
    df = pd.DataFrame({'A': [1, 2, 3, 4, 5], 'B': [10, 20, 30, 40, 50]})
df2 = pd.DataFrame({'A': [6], 'B': [60]})
print 'df\n', df
print 'df2\n', df2

df_x = [df, df2]
result = pd.concat(df_x)
print 'first result\n', result
 
上面代码把df和df2合并为一个result,但是result的index是乱的。

df4.PNG

 
那么执行
result2= result.reset_index()

得到如下的result2: (默认只是返回一个copy,原来的result没有发生改变,所以需要副本赋值给result2)

df5.PNG

可以看到,原来的一列index现在变成了columns之一,新的index为[0,1,2,3,4,5]
如果添加参数 reset_index(drop=True) 那么原index会被丢弃,不会显示为一个新列。
result2 = result.reset_index(drop=True)
df6.PNG

 
reindex的作用是按照原有的列进行重新生成一个新的df。
 
还是使用上面的代码
result目前是df和df2的合并序列。
如下:
df7.PNG

 
可以看到index为[0,1,2,3,4,0]
执行 
result3 = result.reindex(columns=['A','C'])

df8.PNG

 
可以看到,原index并没有发生改变,而列变成了A和C,因为C是不存在的,所以使用了NaB填充,这个值的内容可以自己填充,可以改为默认填充0或者任意你想要的数据。reindex(columns=..)的作用类似于重新把列的顺序整理一遍, 而使用reindex(index=....) 则按照行重新整理一遍。

原文链接:http://30daydo.com/article/257 
欢迎转载,注明出处
 

python 获取 中国证券网 的公告

python爬虫李魔佛 发表了文章 • 11 个评论 • 19421 次浏览 • 2016-06-30 15:45 • 来自相关话题

中国证券网: http://ggjd.cnstock.com/
这个网站的公告会比同花顺东方财富的早一点,而且还出现过早上中国证券网已经发了公告,而东财却拿去做午间公告,以至于可以提前获取公告提前埋伏。
 
现在程序自动把抓取的公告存入本网站中:http://30daydo.com/news.php 
每天早上8:30更新一次。
 
生成的公告保存在stock/文件夹下,以日期命名。 下面脚本是循坏检测,如果有新的公告就会继续生成。
 
默认保存前3页的公告。(一次过太多页会被网站暂时屏蔽几分钟)。 代码以及使用了切换header来躲避网站的封杀。
 
修改
getInfo(3) 里面的数字就可以抓取前面某页数据
 
 




__author__ = 'rocchen'
# working v1.0
from bs4 import BeautifulSoup
import urllib2, datetime, time, codecs, cookielib, random, threading
import os,sys


def getInfo(max_index_user=5):
stock_news_site =
"http://ggjd.cnstock.com/gglist/search/ggkx/"

my_userAgent = [
'Mozilla/5.0 (Macintosh; U; Intel Mac OS X 10_6_8; en-us) AppleWebKit/534.50 (KHTML, like Gecko) Version/5.1 Safari/534.50',
'Mozilla/5.0 (Windows; U; Windows NT 6.1; en-us) AppleWebKit/534.50 (KHTML, like Gecko) Version/5.1 Safari/534.50',
'Mozilla/5.0 (compatible; MSIE 9.0; Windows NT 6.1; Trident/5.0',
'Mozilla/4.0 (compatible; MSIE 8.0; Windows NT 6.0; Trident/4.0)',
'Mozilla/4.0 (compatible; MSIE 7.0; Windows NT 6.0)',
'Mozilla/4.0 (compatible; MSIE 6.0; Windows NT 5.1)',
'Mozilla/5.0 (Macintosh; Intel Mac OS X 10.6; rv:2.0.1) Gecko/20100101 Firefox/4.0.1',
'Mozilla/5.0 (Windows NT 6.1; rv:2.0.1) Gecko/20100101 Firefox/4.0.1',
'Opera/9.80 (Macintosh; Intel Mac OS X 10.6.8; U; en) Presto/2.8.131 Version/11.11',
'Opera/9.80 (Windows NT 6.1; U; en) Presto/2.8.131 Version/11.11',
'Mozilla/4.0 (compatible; MSIE 7.0; Windows NT 5.1; Maxthon 2.0)',
'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_7_0) AppleWebKit/535.11 (KHTML, like Gecko) Chrome/17.0.963.56 Safari/535.11',
'Mozilla/4.0 (compatible; MSIE 7.0; Windows NT 5.1; 360SE)',
'Mozilla/4.0 (compatible; MSIE 7.0; Windows NT 5.1; Trident/4.0; SE 2.X MetaSr 1.0; SE 2.X MetaSr 1.0; .NET CLR 2.0.50727; SE 2.X MetaSr 1.0)']
index = 0
max_index = max_index_user
num = 1
temp_time = time.strftime("[%Y-%m-%d]-[%H-%M]", time.localtime())

store_filename = "StockNews-%s.log" % temp_time
fOpen = codecs.open(store_filename, 'w', 'utf-8')

while index < max_index:
user_agent = random.choice(my_userAgent)
# print user_agent
company_news_site = stock_news_site + str(index)
# content = urllib2.urlopen(company_news_site)
headers = {'User-Agent': user_agent, 'Host': "ggjd.cnstock.com", 'DNT': '1',
'Accept': 'text/html, application/xhtml+xml, */*', }
req = urllib2.Request(url=company_news_site, headers=headers)
resp = None
raw_content = ""
try:
resp = urllib2.urlopen(req, timeout=30)

except urllib2.HTTPError as e:
e.fp.read()
except urllib2.URLError as e:
if hasattr(e, 'code'):
print "error code %d" % e.code
elif hasattr(e, 'reason'):
print "error reason %s " % e.reason

finally:
if resp:
raw_content = resp.read()
time.sleep(2)
resp.close()

soup = BeautifulSoup(raw_content, "html.parser")
all_content = soup.find_all("span", "time")

for i in all_content:
news_time = i.string
node = i.next_sibling
str_temp = "No.%s \n%s\t%s\n---> %s \n\n" % (str(num), news_time, node['title'], node['href'])
#print "inside %d" %num
#print str_temp
fOpen.write(str_temp)
num = num + 1

#print "index %d" %index
index = index + 1

fOpen.close()


def execute_task(n=60):
period = int(n)
while True:
print datetime.datetime.now()
getInfo(3)

time.sleep(60 * period)



if __name__ == "__main__":

sub_folder = os.path.join(os.getcwd(), "stock")
if not os.path.exists(sub_folder):
os.mkdir(sub_folder)
os.chdir(sub_folder)
start_time = time.time() # user can change the max index number getInfo(10), by default is getInfo(5)
if len(sys.argv) <2:
n = raw_input("Input Period : ? mins to download every cycle")
else:
n=int(sys.argv[1])
execute_task(n)
end_time = time.time()
print "Total time: %s s." % str(round((end_time - start_time), 4))


 
github:https://github.com/Rockyzsu/cnstock
  查看全部
中国证券网: http://ggjd.cnstock.com/
这个网站的公告会比同花顺东方财富的早一点,而且还出现过早上中国证券网已经发了公告,而东财却拿去做午间公告,以至于可以提前获取公告提前埋伏。
 
现在程序自动把抓取的公告存入本网站中:http://30daydo.com/news.php 
每天早上8:30更新一次。
 
生成的公告保存在stock/文件夹下,以日期命名。 下面脚本是循坏检测,如果有新的公告就会继续生成。
 
默认保存前3页的公告。(一次过太多页会被网站暂时屏蔽几分钟)。 代码以及使用了切换header来躲避网站的封杀。
 
修改
getInfo(3) 里面的数字就可以抓取前面某页数据
 
 

公告.PNG
__author__ = 'rocchen'
# working v1.0
from bs4 import BeautifulSoup
import urllib2, datetime, time, codecs, cookielib, random, threading
import os,sys


def getInfo(max_index_user=5):
stock_news_site =
"http://ggjd.cnstock.com/gglist/search/ggkx/"

my_userAgent = [
'Mozilla/5.0 (Macintosh; U; Intel Mac OS X 10_6_8; en-us) AppleWebKit/534.50 (KHTML, like Gecko) Version/5.1 Safari/534.50',
'Mozilla/5.0 (Windows; U; Windows NT 6.1; en-us) AppleWebKit/534.50 (KHTML, like Gecko) Version/5.1 Safari/534.50',
'Mozilla/5.0 (compatible; MSIE 9.0; Windows NT 6.1; Trident/5.0',
'Mozilla/4.0 (compatible; MSIE 8.0; Windows NT 6.0; Trident/4.0)',
'Mozilla/4.0 (compatible; MSIE 7.0; Windows NT 6.0)',
'Mozilla/4.0 (compatible; MSIE 6.0; Windows NT 5.1)',
'Mozilla/5.0 (Macintosh; Intel Mac OS X 10.6; rv:2.0.1) Gecko/20100101 Firefox/4.0.1',
'Mozilla/5.0 (Windows NT 6.1; rv:2.0.1) Gecko/20100101 Firefox/4.0.1',
'Opera/9.80 (Macintosh; Intel Mac OS X 10.6.8; U; en) Presto/2.8.131 Version/11.11',
'Opera/9.80 (Windows NT 6.1; U; en) Presto/2.8.131 Version/11.11',
'Mozilla/4.0 (compatible; MSIE 7.0; Windows NT 5.1; Maxthon 2.0)',
'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_7_0) AppleWebKit/535.11 (KHTML, like Gecko) Chrome/17.0.963.56 Safari/535.11',
'Mozilla/4.0 (compatible; MSIE 7.0; Windows NT 5.1; 360SE)',
'Mozilla/4.0 (compatible; MSIE 7.0; Windows NT 5.1; Trident/4.0; SE 2.X MetaSr 1.0; SE 2.X MetaSr 1.0; .NET CLR 2.0.50727; SE 2.X MetaSr 1.0)']
index = 0
max_index = max_index_user
num = 1
temp_time = time.strftime("[%Y-%m-%d]-[%H-%M]", time.localtime())

store_filename = "StockNews-%s.log" % temp_time
fOpen = codecs.open(store_filename, 'w', 'utf-8')

while index < max_index:
user_agent = random.choice(my_userAgent)
# print user_agent
company_news_site = stock_news_site + str(index)
# content = urllib2.urlopen(company_news_site)
headers = {'User-Agent': user_agent, 'Host': "ggjd.cnstock.com", 'DNT': '1',
'Accept': 'text/html, application/xhtml+xml, */*', }
req = urllib2.Request(url=company_news_site, headers=headers)
resp = None
raw_content = ""
try:
resp = urllib2.urlopen(req, timeout=30)

except urllib2.HTTPError as e:
e.fp.read()
except urllib2.URLError as e:
if hasattr(e, 'code'):
print "error code %d" % e.code
elif hasattr(e, 'reason'):
print "error reason %s " % e.reason

finally:
if resp:
raw_content = resp.read()
time.sleep(2)
resp.close()

soup = BeautifulSoup(raw_content, "html.parser")
all_content = soup.find_all("span", "time")

for i in all_content:
news_time = i.string
node = i.next_sibling
str_temp = "No.%s \n%s\t%s\n---> %s \n\n" % (str(num), news_time, node['title'], node['href'])
#print "inside %d" %num
#print str_temp
fOpen.write(str_temp)
num = num + 1

#print "index %d" %index
index = index + 1

fOpen.close()


def execute_task(n=60):
period = int(n)
while True:
print datetime.datetime.now()
getInfo(3)

time.sleep(60 * period)



if __name__ == "__main__":

sub_folder = os.path.join(os.getcwd(), "stock")
if not os.path.exists(sub_folder):
os.mkdir(sub_folder)
os.chdir(sub_folder)
start_time = time.time() # user can change the max index number getInfo(10), by default is getInfo(5)
if len(sys.argv) <2:
n = raw_input("Input Period : ? mins to download every cycle")
else:
n=int(sys.argv[1])
execute_task(n)
end_time = time.time()
print "Total time: %s s." % str(round((end_time - start_time), 4))


 
github:https://github.com/Rockyzsu/cnstock
 

python 批量获取色影无忌 获奖图片

python爬虫李魔佛 发表了文章 • 6 个评论 • 14297 次浏览 • 2016-06-29 16:41 • 来自相关话题

色影无忌上的图片很多都可以直接拿来做壁纸的,而且发布面不会太广,基本不会和市面上大部分的壁纸或者图片素材重复。 关键还没有水印。 这么良心的图片服务商哪里找呀~~
 

 





 
不多说,直接来代码:#-*-coding=utf-8-*-
__author__ = 'rocky chen'
from bs4 import BeautifulSoup
import urllib2,sys,StringIO,gzip,time,random,re,urllib,os
reload(sys)
sys.setdefaultencoding('utf-8')
class Xitek():
    def __init__(self):
        self.url="http://photo.xitek.com/"
        user_agent="Mozilla/5.0 (compatible; MSIE 9.0; Windows NT 6.1; Trident/5.0)"
        self.headers={"User-Agent":user_agent}
        self.last_page=self.__get_last_page()


    def __get_last_page(self):
        html=self.__getContentAuto(self.url)
        bs=BeautifulSoup(html,"html.parser")
        page=bs.find_all('a',class_="blast")
        last_page=page[0]['href'].split('/')[-1]
        return int(last_page)


    def __getContentAuto(self,url):
        req=urllib2.Request(url,headers=self.headers)
        resp=urllib2.urlopen(req)
        #time.sleep(2*random.random())
        content=resp.read()
        info=resp.info().get("Content-Encoding")
        if info==None:
            return content
        else:
            t=StringIO.StringIO(content)
            gziper=gzip.GzipFile(fileobj=t)
            html = gziper.read()
            return html

    #def __getFileName(self,stream):


    def __download(self,url):
        p=re.compile(r'href="(/photoid/\d+)"')
        #html=self.__getContentNoZip(url)

        html=self.__getContentAuto(url)

        content = p.findall(html)
        for i in content:
            print i

            photoid=self.__getContentAuto(self.url+i)
            bs=BeautifulSoup(photoid,"html.parser")
            final_link=bs.find('img',class_="mimg")['src']
            print final_link
            #pic_stream=self.__getContentAuto(final_link)
            title=bs.title.string.strip()
            filename = re.sub('[\/:*?"<>|]', '-', title)
            filename=filename+'.jpg'
            urllib.urlretrieve(final_link,filename)
            #f=open(filename,'w')
            #f.write(pic_stream)
            #f.close()
        #print html
        #bs=BeautifulSoup(html,"html.parser")
        #content=bs.find_all(p)
        #for i in content:
        #    print i
        '''
        print bs.title
        element_link=bs.find_all('div',class_="element")
        print len(element_link)
        k=1
        for href in element_link:

            #print type(href)
            #print href.tag
        '''
        '''
            if href.children[0]:
                print href.children[0]
        '''
        '''
            t=0

            for i in href.children:
                #if i.a:
                if t==0:
                    #print k
                    if i['href']
                    print link

                        if p.findall(link):
                            full_path=self.url[0:len(self.url)-1]+link
                            sub_html=self.__getContent(full_path)
                            bs=BeautifulSoup(sub_html,"html.parser")
                            final_link=bs.find('img',class_="mimg")['src']
                            #time.sleep(2*random.random())
                            print final_link
                    #k=k+1
                #print type(i)
                #print i.tag
                #if hasattr(i,"href"):
                    #print i['href']
                #print i.tag
                t=t+1
                #print "*"

        '''

        '''
            if href:
                if href.children:
                    print href.children[0]
        '''
            #print "one element link"



    def getPhoto(self):

        start=0
        #use style/0
        photo_url="http://photo.xitek.com/style/0/p/"
        for i in range(start,self.last_page+1):
            url=photo_url+str(i)
            print url
            #time.sleep(1)
            self.__download(url)

        '''
        url="http://photo.xitek.com/style/0/p/10"
        self.__download(url)
        '''
        #url="http://photo.xitek.com/style/0/p/0"
        #html=self.__getContent(url)
        #url="http://photo.xitek.com/"
        #html=self.__getContentNoZip(url)
        #print html
        #'''
def main():
    sub_folder = os.path.join(os.getcwd(), "content")
    if not os.path.exists(sub_folder):
        os.mkdir(sub_folder)
    os.chdir(sub_folder)
    obj=Xitek()
    obj.getPhoto()


if __name__=="__main__":
    main()








下载后在content文件夹下会自动抓取所有图片。 (色影无忌的服务器没有做任何的屏蔽处理,所以脚本不能跑那么快,可以适当调用sleep函数,不要让服务器压力那么大)
 
已经下载好的图片:





 
 
github: https://github.com/Rockyzsu/fetchXitek   (欢迎前来star) 查看全部
色影无忌上的图片很多都可以直接拿来做壁纸的,而且发布面不会太广,基本不会和市面上大部分的壁纸或者图片素材重复。 关键还没有水印。 这么良心的图片服务商哪里找呀~~
 

 

色影无忌_副本.png

 
不多说,直接来代码:
#-*-coding=utf-8-*-
__author__ = 'rocky chen'
from bs4 import BeautifulSoup
import urllib2,sys,StringIO,gzip,time,random,re,urllib,os
reload(sys)
sys.setdefaultencoding('utf-8')
class Xitek():
    def __init__(self):
        self.url="http://photo.xitek.com/"
        user_agent="Mozilla/5.0 (compatible; MSIE 9.0; Windows NT 6.1; Trident/5.0)"
        self.headers={"User-Agent":user_agent}
        self.last_page=self.__get_last_page()


    def __get_last_page(self):
        html=self.__getContentAuto(self.url)
        bs=BeautifulSoup(html,"html.parser")
        page=bs.find_all('a',class_="blast")
        last_page=page[0]['href'].split('/')[-1]
        return int(last_page)


    def __getContentAuto(self,url):
        req=urllib2.Request(url,headers=self.headers)
        resp=urllib2.urlopen(req)
        #time.sleep(2*random.random())
        content=resp.read()
        info=resp.info().get("Content-Encoding")
        if info==None:
            return content
        else:
            t=StringIO.StringIO(content)
            gziper=gzip.GzipFile(fileobj=t)
            html = gziper.read()
            return html

    #def __getFileName(self,stream):


    def __download(self,url):
        p=re.compile(r'href="(/photoid/\d+)"')
        #html=self.__getContentNoZip(url)

        html=self.__getContentAuto(url)

        content = p.findall(html)
        for i in content:
            print i

            photoid=self.__getContentAuto(self.url+i)
            bs=BeautifulSoup(photoid,"html.parser")
            final_link=bs.find('img',class_="mimg")['src']
            print final_link
            #pic_stream=self.__getContentAuto(final_link)
            title=bs.title.string.strip()
            filename = re.sub('[\/:*?"<>|]', '-', title)
            filename=filename+'.jpg'
            urllib.urlretrieve(final_link,filename)
            #f=open(filename,'w')
            #f.write(pic_stream)
            #f.close()
        #print html
        #bs=BeautifulSoup(html,"html.parser")
        #content=bs.find_all(p)
        #for i in content:
        #    print i
        '''
        print bs.title
        element_link=bs.find_all('div',class_="element")
        print len(element_link)
        k=1
        for href in element_link:

            #print type(href)
            #print href.tag
        '''
        '''
            if href.children[0]:
                print href.children[0]
        '''
        '''
            t=0

            for i in href.children:
                #if i.a:
                if t==0:
                    #print k
                    if i['href']
                    print link

                        if p.findall(link):
                            full_path=self.url[0:len(self.url)-1]+link
                            sub_html=self.__getContent(full_path)
                            bs=BeautifulSoup(sub_html,"html.parser")
                            final_link=bs.find('img',class_="mimg")['src']
                            #time.sleep(2*random.random())
                            print final_link
                    #k=k+1
                #print type(i)
                #print i.tag
                #if hasattr(i,"href"):
                    #print i['href']
                #print i.tag
                t=t+1
                #print "*"

        '''

        '''
            if href:
                if href.children:
                    print href.children[0]
        '''
            #print "one element link"



    def getPhoto(self):

        start=0
        #use style/0
        photo_url="http://photo.xitek.com/style/0/p/"
        for i in range(start,self.last_page+1):
            url=photo_url+str(i)
            print url
            #time.sleep(1)
            self.__download(url)

        '''
        url="http://photo.xitek.com/style/0/p/10"
        self.__download(url)
        '''
        #url="http://photo.xitek.com/style/0/p/0"
        #html=self.__getContent(url)
        #url="http://photo.xitek.com/"
        #html=self.__getContentNoZip(url)
        #print html
        #'''
def main():
    sub_folder = os.path.join(os.getcwd(), "content")
    if not os.path.exists(sub_folder):
        os.mkdir(sub_folder)
    os.chdir(sub_folder)
    obj=Xitek()
    obj.getPhoto()


if __name__=="__main__":
    main()








下载后在content文件夹下会自动抓取所有图片。 (色影无忌的服务器没有做任何的屏蔽处理,所以脚本不能跑那么快,可以适当调用sleep函数,不要让服务器压力那么大)
 
已经下载好的图片:

色影无忌2_副本1.png

 
 
github: https://github.com/Rockyzsu/fetchXitek   (欢迎前来star)

抓取 知乎日报 中的 大误 系类文章,生成电子书推送到kindle

python爬虫李魔佛 发表了文章 • 0 个评论 • 7429 次浏览 • 2016-06-12 08:52 • 来自相关话题

无意中看了知乎日报的大误系列的一篇文章,之后就停不下来了,大误是虚构故事,知乎上神人虚构故事的功力要高于网络上的很多写手啊!! 看的欲罢不能,不过还是那句,手机屏幕太小,连续看几个小时很疲劳,而且每次都要联网去看。 
 
所以写了下面的python脚本,一劳永逸。 脚本抓取大误从开始到现在的所有文章,并推送到你自己的kindle账号。
 




# -*- coding=utf-8 -*-
__author__ = 'rocky @ www.30daydo.com'
import urllib2, re, os, codecs,sys,datetime
from bs4 import BeautifulSoup
# example https://zhhrb.sinaapp.com/index.php?date=20160610
from mail_template import MailAtt
reload(sys)
sys.setdefaultencoding('utf-8')

def save2file(filename, content):
filename = filename + ".txt"
f = codecs.open(filename, 'a', encoding='utf-8')
f.write(content)
f.close()


def getPost(date_time, filter_p):
url = 'https://zhhrb.sinaapp.com/index.php?date=' + date_time
user_agent = "Mozilla/5.0 (compatible; MSIE 9.0; Windows NT 6.1; Trident/5.0)"
header = {"User-Agent": user_agent}
req = urllib2.Request(url, headers=header)
resp = urllib2.urlopen(req)
content = resp.read()
p = re.compile('<h2 class="question-title">(.*)</h2></br></a>')
result = re.findall(p, content)
count = -1
row = -1
for i in result:
#print i
return_content = re.findall(filter_p, i)

if return_content:
row = count
break
#print return_content[0]
count = count + 1
#print row
if row == -1:
return 0
link_p = re.compile('<a href="(.*)" target="_blank" rel="nofollow">')
link_result = re.findall(link_p, content)[row + 1]
print link_result
result_req = urllib2.Request(link_result, headers=header)
result_resp = urllib2.urlopen(result_req)
#result_content= result_resp.read()
#print result_content

bs = BeautifulSoup(result_resp, "html.parser")
title = bs.title.string.strip()
#print title
filename = re.sub('[\/:*?"<>|]', '-', title)
print filename
print date_time
save2file(filename, title)
save2file(filename, "\n\n\n\n--------------------%s Detail----------------------\n\n" %date_time)

detail_content = bs.find_all('div', class_='content')

for i in detail_content:
#print i
save2file(filename,"\n\n-------------------------answer -------------------------\n\n")
for j in i.strings:

save2file(filename, j)

smtp_server = 'smtp.126.com'
from_mail = sys.argv[1]
password = sys.argv[2]
to_mail = 'xxxxx@kindle.cn'
send_kindle = MailAtt(smtp_server, from_mail, password, to_mail)
send_kindle.send_txt(filename)


def main():
sub_folder = os.path.join(os.getcwd(), "content")
if not os.path.exists(sub_folder):
os.mkdir(sub_folder)
os.chdir(sub_folder)


date_time = '20160611'
filter_p = re.compile('大误.*')
ori_day=datetime.date(datetime.date.today().year,01,01)
t=datetime.date(datetime.date.today().year,datetime.date.today().month,datetime.date.today().day)
delta=(t-ori_day).days
print delta
for i in range(delta):
day=datetime.date(datetime.date.today().year,01,01)+datetime.timedelta(i)
getPost(day.strftime("%Y%m%d"),filter_p)
#getPost(date_time, filter_p)

if __name__ == "__main__":
main()





github: https://github.com/Rockyzsu/zhihu_daily__kindle
 
上面的代码可以稍作修改,就可以抓取瞎扯或者深夜食堂的系列文章。
 
附福利:
http://pan.baidu.com/s/1kVewz59
所有的知乎日报的大误文章。(截止2016/6/12日) 查看全部
无意中看了知乎日报的大误系列的一篇文章,之后就停不下来了,大误是虚构故事,知乎上神人虚构故事的功力要高于网络上的很多写手啊!! 看的欲罢不能,不过还是那句,手机屏幕太小,连续看几个小时很疲劳,而且每次都要联网去看。 
 
所以写了下面的python脚本,一劳永逸。 脚本抓取大误从开始到现在的所有文章,并推送到你自己的kindle账号。
 

大误.JPG
# -*- coding=utf-8 -*-
__author__ = 'rocky @ www.30daydo.com'
import urllib2, re, os, codecs,sys,datetime
from bs4 import BeautifulSoup
# example https://zhhrb.sinaapp.com/index.php?date=20160610
from mail_template import MailAtt
reload(sys)
sys.setdefaultencoding('utf-8')

def save2file(filename, content):
filename = filename + ".txt"
f = codecs.open(filename, 'a', encoding='utf-8')
f.write(content)
f.close()


def getPost(date_time, filter_p):
url = 'https://zhhrb.sinaapp.com/index.php?date=' + date_time
user_agent = "Mozilla/5.0 (compatible; MSIE 9.0; Windows NT 6.1; Trident/5.0)"
header = {"User-Agent": user_agent}
req = urllib2.Request(url, headers=header)
resp = urllib2.urlopen(req)
content = resp.read()
p = re.compile('<h2 class="question-title">(.*)</h2></br></a>')
result = re.findall(p, content)
count = -1
row = -1
for i in result:
#print i
return_content = re.findall(filter_p, i)

if return_content:
row = count
break
#print return_content[0]
count = count + 1
#print row
if row == -1:
return 0
link_p = re.compile('<a href="(.*)" target="_blank" rel="nofollow">')
link_result = re.findall(link_p, content)[row + 1]
print link_result
result_req = urllib2.Request(link_result, headers=header)
result_resp = urllib2.urlopen(result_req)
#result_content= result_resp.read()
#print result_content

bs = BeautifulSoup(result_resp, "html.parser")
title = bs.title.string.strip()
#print title
filename = re.sub('[\/:*?"<>|]', '-', title)
print filename
print date_time
save2file(filename, title)
save2file(filename, "\n\n\n\n--------------------%s Detail----------------------\n\n" %date_time)

detail_content = bs.find_all('div', class_='content')

for i in detail_content:
#print i
save2file(filename,"\n\n-------------------------answer -------------------------\n\n")
for j in i.strings:

save2file(filename, j)

smtp_server = 'smtp.126.com'
from_mail = sys.argv[1]
password = sys.argv[2]
to_mail = 'xxxxx@kindle.cn'
send_kindle = MailAtt(smtp_server, from_mail, password, to_mail)
send_kindle.send_txt(filename)


def main():
sub_folder = os.path.join(os.getcwd(), "content")
if not os.path.exists(sub_folder):
os.mkdir(sub_folder)
os.chdir(sub_folder)


date_time = '20160611'
filter_p = re.compile('大误.*')
ori_day=datetime.date(datetime.date.today().year,01,01)
t=datetime.date(datetime.date.today().year,datetime.date.today().month,datetime.date.today().day)
delta=(t-ori_day).days
print delta
for i in range(delta):
day=datetime.date(datetime.date.today().year,01,01)+datetime.timedelta(i)
getPost(day.strftime("%Y%m%d"),filter_p)
#getPost(date_time, filter_p)

if __name__ == "__main__":
main()





github: https://github.com/Rockyzsu/zhihu_daily__kindle
 
上面的代码可以稍作修改,就可以抓取瞎扯或者深夜食堂的系列文章。
 
附福利:
http://pan.baidu.com/s/1kVewz59
所有的知乎日报的大误文章。(截止2016/6/12日)

python 爆解zip压缩文件密码

python李魔佛 发表了文章 • 0 个评论 • 7509 次浏览 • 2016-06-09 21:43 • 来自相关话题

出于对百度网盘的不信任,加上前阵子百度会把一些侵犯版权的文件清理掉或者一些百度认为的尺度过大的文件进行替换,留下一个4秒的教育视频。 为何不提前告诉用户? 擅自把用户的资料删除,以后用户哪敢随意把资料上传上去呢?
 
抱怨归抱怨,由于现在金山快盘,新浪尾盘都关闭了,速度稍微快点的就只有百度网盘了。 所以我会把文件事先压缩好,加个密码然后上传。
 
可是有时候下载下来却忘记了解压密码,实在蛋疼。 所以需要自己逐一验证密码。 所以就写了这个小脚本。 很简单,没啥技术含量。 
 





 
 
代码就用图片吧,大家可以上机自己敲敲代码也好。 ctrl+v 代码 其实会养成一种惰性。
 
github: https://github.com/Rockyzsu/zip_crash
  查看全部
出于对百度网盘的不信任,加上前阵子百度会把一些侵犯版权的文件清理掉或者一些百度认为的尺度过大的文件进行替换,留下一个4秒的教育视频。 为何不提前告诉用户? 擅自把用户的资料删除,以后用户哪敢随意把资料上传上去呢?
 
抱怨归抱怨,由于现在金山快盘,新浪尾盘都关闭了,速度稍微快点的就只有百度网盘了。 所以我会把文件事先压缩好,加个密码然后上传。
 
可是有时候下载下来却忘记了解压密码,实在蛋疼。 所以需要自己逐一验证密码。 所以就写了这个小脚本。 很简单,没啥技术含量。 
 

crash_zip.JPG

 
 
代码就用图片吧,大家可以上机自己敲敲代码也好。 ctrl+v 代码 其实会养成一种惰性。
 
github: https://github.com/Rockyzsu/zip_crash
 

python雪球爬虫 抓取雪球 大V的所有文章 推送到kindle

python爬虫李魔佛 发表了文章 • 3 个评论 • 17829 次浏览 • 2016-05-29 00:06 • 来自相关话题

30天内完成。 开始日期:2016年5月28日
 
因为雪球上喷子很多,不少大V都不堪忍受,被喷的删帖离开。 比如 易碎品,小小辛巴。
所以利用python可以有效便捷的抓取想要的大V发言内容,并保存到本地。也方便自己检索,考证(有些伪大V喜欢频繁删帖,比如今天预测明天大盘大涨,明天暴跌后就把昨天的预测给删掉,给后来者造成的错觉改大V每次都能精准预测)。 
 
下面以 抓取狂龙的帖子为例(狂龙最近老是掀人家庄家的老底,哈)
 
https://xueqiu.com/4742988362 
 
2017年2月20日更新:
爬取雪球上我的收藏的文章,并生成电子书。
(PS:收藏夹中一些文章已经被作者删掉了 - -|, 这速度也蛮快了呀。估计是以前写的现在怕被放出来打脸)
 




# -*-coding=utf-8-*-
#抓取雪球的收藏文章
__author__ = 'Rocky'
import requests,cookielib,re,json,time
from toolkit import Toolkit
from lxml import etree
url='https://xueqiu.com/snowman/login'
session = requests.session()

session.cookies = cookielib.LWPCookieJar(filename="cookies")
try:
session.cookies.load(ignore_discard=True)
except:
print "Cookie can't load"

agent = 'Mozilla/5.0 (Windows NT 5.1; rv:33.0) Gecko/20100101 Firefox/33.0'
headers = {'Host': 'xueqiu.com',
'Referer': 'https://xueqiu.com/',
'Origin':'https://xueqiu.com',
'User-Agent': agent}
account=Toolkit.getUserData('data.cfg')
print account['snowball_user']
print account['snowball_password']

data={'username':account['snowball_user'],'password':account['snowball_password']}
s=session.post(url,data=data,headers=headers)
print s.status_code
#print s.text
session.cookies.save()
fav_temp='https://xueqiu.com/favs?page=1'
collection=session.get(fav_temp,headers=headers)
fav_content= collection.text
p=re.compile('"maxPage":(\d+)')
maxPage=p.findall(fav_content)[0]
print maxPage
print type(maxPage)
maxPage=int(maxPage)
print type(maxPage)
for i in range(1,maxPage+1):
fav='https://xueqiu.com/favs?page=%d' %i
collection=session.get(fav,headers=headers)
fav_content= collection.text
#print fav_content
p=re.compile('var favs = {(.*?)};',re.S|re.M)
result=p.findall(fav_content)[0].strip()

new_result='{'+result+'}'
#print type(new_result)
#print new_result
data=json.loads(new_result)
use_data= data['list']
host='https://xueqiu.com'
for i in use_data:
url=host+ i['target']
print url
txt_content=session.get(url,headers=headers).text
#print txt_content.text

tree=etree.HTML(txt_content)
title=tree.xpath('//title/text()')[0]

filename = re.sub('[\/:*?"<>|]', '-', title)
print filename

content=tree.xpath('//div[@class="detail"]')
for i in content:
Toolkit.save2filecn(filename, i.xpath('string(.)'))
#print content
#Toolkit.save2file(filename,)
time.sleep(10)





 
用法:
1. snowball.py -- 抓取雪球上我的收藏的文章
使用: 创建一个data.cfg的文件,里面格式如下:
snowball_user=xxxxx@xx.com
snowball_password=密码

然后运行python snowball.py ,会自动登录雪球,然后 在当前目录生产txt文件。
 
github代码:https://github.com/Rockyzsu/xueqiu 查看全部
30天内完成。 开始日期:2016年5月28日
 
因为雪球上喷子很多,不少大V都不堪忍受,被喷的删帖离开。 比如 易碎品,小小辛巴。
所以利用python可以有效便捷的抓取想要的大V发言内容,并保存到本地。也方便自己检索,考证(有些伪大V喜欢频繁删帖,比如今天预测明天大盘大涨,明天暴跌后就把昨天的预测给删掉,给后来者造成的错觉改大V每次都能精准预测)。 
 
下面以 抓取狂龙的帖子为例(狂龙最近老是掀人家庄家的老底,哈)
 
https://xueqiu.com/4742988362 
 
2017年2月20日更新:
爬取雪球上我的收藏的文章,并生成电子书。
(PS:收藏夹中一些文章已经被作者删掉了 - -|, 这速度也蛮快了呀。估计是以前写的现在怕被放出来打脸)
 

雪球的爬虫.PNG
# -*-coding=utf-8-*-
#抓取雪球的收藏文章
__author__ = 'Rocky'
import requests,cookielib,re,json,time
from toolkit import Toolkit
from lxml import etree
url='https://xueqiu.com/snowman/login'
session = requests.session()

session.cookies = cookielib.LWPCookieJar(filename="cookies")
try:
session.cookies.load(ignore_discard=True)
except:
print "Cookie can't load"

agent = 'Mozilla/5.0 (Windows NT 5.1; rv:33.0) Gecko/20100101 Firefox/33.0'
headers = {'Host': 'xueqiu.com',
'Referer': 'https://xueqiu.com/',
'Origin':'https://xueqiu.com',
'User-Agent': agent}
account=Toolkit.getUserData('data.cfg')
print account['snowball_user']
print account['snowball_password']

data={'username':account['snowball_user'],'password':account['snowball_password']}
s=session.post(url,data=data,headers=headers)
print s.status_code
#print s.text
session.cookies.save()
fav_temp='https://xueqiu.com/favs?page=1'
collection=session.get(fav_temp,headers=headers)
fav_content= collection.text
p=re.compile('"maxPage":(\d+)')
maxPage=p.findall(fav_content)[0]
print maxPage
print type(maxPage)
maxPage=int(maxPage)
print type(maxPage)
for i in range(1,maxPage+1):
fav='https://xueqiu.com/favs?page=%d' %i
collection=session.get(fav,headers=headers)
fav_content= collection.text
#print fav_content
p=re.compile('var favs = {(.*?)};',re.S|re.M)
result=p.findall(fav_content)[0].strip()

new_result='{'+result+'}'
#print type(new_result)
#print new_result
data=json.loads(new_result)
use_data= data['list']
host='https://xueqiu.com'
for i in use_data:
url=host+ i['target']
print url
txt_content=session.get(url,headers=headers).text
#print txt_content.text

tree=etree.HTML(txt_content)
title=tree.xpath('//title/text()')[0]

filename = re.sub('[\/:*?"<>|]', '-', title)
print filename

content=tree.xpath('//div[@class="detail"]')
for i in content:
Toolkit.save2filecn(filename, i.xpath('string(.)'))
#print content
#Toolkit.save2file(filename,)
time.sleep(10)





 
用法:
1. snowball.py -- 抓取雪球上我的收藏的文章
使用: 创建一个data.cfg的文件,里面格式如下:
snowball_user=xxxxx@xx.com
snowball_password=密码

然后运行python snowball.py ,会自动登录雪球,然后 在当前目录生产txt文件。
 
github代码:https://github.com/Rockyzsu/xueqiu

python 多线程扫描开放端口

python低调的哥哥 发表了文章 • 0 个评论 • 8642 次浏览 • 2016-05-15 21:15 • 来自相关话题

为什么说python是黑客的语言? 因为很多扫描+破解的任务都可以用python很快的实现,简洁明了。且有大量的库来支持。import socket,sys
import time
from thread_test import MyThread

socket.setdefaulttimeout(1)
#设置每个线程socket的timeou时间,超过1秒没有反应就认为端口不开放
thread_num=4
#线程数目
ip_end=256
ip_start=0
scope=ip_end/thread_num

def scan(ip_head,ip_low, port):
try:
# Alert !!! below statement should be inside scan function. Else each it is one s
ip=ip_head+str(ip_low)
print ip
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
s.connect((ip, port))
#通过这一句判断 是否连通
s.close()
print "ip %s port %d open\n" %(ip,port)
return True
except:
return False


def scan_range(ip_head,ip_range,port):
start,end=ip_range
for i in range(start,end):
scan(ip_head,i,port)

if len(sys.argv)<3:
print "input ip and port"
exit()

ip_head=sys.argv[1]
port=int(sys.argv[2])


ip_range=
for i in range(thread_num):
x_range=[i*scope,(i+1)*scope-1]
ip_range.append(x_range)

threads=
for i in range(thread_num):
t=MyThread(scan_range,(ip_head,ip_range,port))
threads.append(t)
for i in range(thread_num):
threads.start()
for i in range(thread_num):
threads.join()
#设置进程阻塞,防止主线程退出了,其他的多线程还在运行

print "*****end*****"多线程的类函数实现: 有一些测试函数在上面没注释或者删除掉,为了让一些初学者更加容易看懂。import thread,threading,time,datetime
from time import sleep,ctime
def loop1():
print "start %s " %ctime()
print "start in loop1"
sleep(3)
print "end %s " %ctime()

def loop2():
print "sart %s " %ctime()
print "start in loop2"
sleep(6)
print "end %s " %ctime()


class MyThread(threading.Thread):
def __init__(self,fun,arg,name=""):
threading.Thread.__init__(self)
self.fun=fun
self.arg=arg
self.name=name
#self.result

def run(self):
self.result=apply(self.fun,self.arg)

def getResult(self):
return self.result

def fib(n):
if n<2:
return 1
else:
return fib(n-1)+fib(n-2)


def sum(n):
if n<2:
return 1
else:
return n+sum(n-1)

def fab(n):
if n<2:
return 1
else:
return n*fab(n-1)

def single_thread():
print fib(12)
print sum(12)
print fab(12)


def multi_thread():
print "in multithread"
fun_list=[fib,sum,fab]
n=len(fun_list)
threads=
count=12
for i in range(n):
t=MyThread(fun_list,(count,),fun_list.__name__)
threads.append(t)
for i in range(n):
threads.start()

for i in range(n):
threads.join()
result= threads.getResult()
print result
def main():
'''
print "start at main"
thread.start_new_thread(loop1,())
thread.start_new_thread(loop2,())
sleep(10)
print "end at main"
'''
start=ctime()
#print "Used %f" %(end-start).seconds
print start
single_thread()
end=ctime()
print end
multi_thread()
#print "used %s" %(end-start).seconds
if __name__=="__main__":
main()
 
最终运行的格式就是  python scan_host.py 192.168.1. 22
上面的命令就是扫描192.168.1 ip段开启了22端口服务的机器,也就是ssh服务。 
 
github:https://github.com/Rockyzsu/scan_host​ 

  查看全部
为什么说python是黑客的语言? 因为很多扫描+破解的任务都可以用python很快的实现,简洁明了。且有大量的库来支持。
import socket,sys
import time
from thread_test import MyThread

socket.setdefaulttimeout(1)
#设置每个线程socket的timeou时间,超过1秒没有反应就认为端口不开放
thread_num=4
#线程数目
ip_end=256
ip_start=0
scope=ip_end/thread_num

def scan(ip_head,ip_low, port):
try:
# Alert !!! below statement should be inside scan function. Else each it is one s
ip=ip_head+str(ip_low)
print ip
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
s.connect((ip, port))
#通过这一句判断 是否连通
s.close()
print "ip %s port %d open\n" %(ip,port)
return True
except:
return False


def scan_range(ip_head,ip_range,port):
start,end=ip_range
for i in range(start,end):
scan(ip_head,i,port)

if len(sys.argv)<3:
print "input ip and port"
exit()

ip_head=sys.argv[1]
port=int(sys.argv[2])


ip_range=
for i in range(thread_num):
x_range=[i*scope,(i+1)*scope-1]
ip_range.append(x_range)

threads=
for i in range(thread_num):
t=MyThread(scan_range,(ip_head,ip_range,port))
threads.append(t)
for i in range(thread_num):
threads.start()
for i in range(thread_num):
threads.join()
#设置进程阻塞,防止主线程退出了,其他的多线程还在运行

print "*****end*****"
多线程的类函数实现: 有一些测试函数在上面没注释或者删除掉,为了让一些初学者更加容易看懂。
import thread,threading,time,datetime
from time import sleep,ctime
def loop1():
print "start %s " %ctime()
print "start in loop1"
sleep(3)
print "end %s " %ctime()

def loop2():
print "sart %s " %ctime()
print "start in loop2"
sleep(6)
print "end %s " %ctime()


class MyThread(threading.Thread):
def __init__(self,fun,arg,name=""):
threading.Thread.__init__(self)
self.fun=fun
self.arg=arg
self.name=name
#self.result

def run(self):
self.result=apply(self.fun,self.arg)

def getResult(self):
return self.result

def fib(n):
if n<2:
return 1
else:
return fib(n-1)+fib(n-2)


def sum(n):
if n<2:
return 1
else:
return n+sum(n-1)

def fab(n):
if n<2:
return 1
else:
return n*fab(n-1)

def single_thread():
print fib(12)
print sum(12)
print fab(12)


def multi_thread():
print "in multithread"
fun_list=[fib,sum,fab]
n=len(fun_list)
threads=
count=12
for i in range(n):
t=MyThread(fun_list,(count,),fun_list.__name__)
threads.append(t)
for i in range(n):
threads.start()

for i in range(n):
threads.join()
result= threads.getResult()
print result
def main():
'''
print "start at main"
thread.start_new_thread(loop1,())
thread.start_new_thread(loop2,())
sleep(10)
print "end at main"
'''
start=ctime()
#print "Used %f" %(end-start).seconds
print start
single_thread()
end=ctime()
print end
multi_thread()
#print "used %s" %(end-start).seconds
if __name__=="__main__":
main()

 
最终运行的格式就是  python scan_host.py 192.168.1. 22
上面的命令就是扫描192.168.1 ip段开启了22端口服务的机器,也就是ssh服务。 
 
github:https://github.com/Rockyzsu/scan_host​ 

 

kindle收不到python推送的附件,但是同邮件的客户端可以。求助。

回复

python李魔佛 回复了问题 • 2 人关注 • 1 个回复 • 2833 次浏览 • 2019-04-08 10:03 • 来自相关话题

RuntimeWarning: More than 20 figures have been opened.

回复

python李魔佛 回复了问题 • 1 人关注 • 1 个回复 • 12238 次浏览 • 2018-04-12 12:40 • 来自相关话题

真像雪球和知乎啊,这种是用python开发的后台吗,是用的什么框架呢

回复

默认分类kflyddn 回复了问题 • 3 人关注 • 3 个回复 • 7101 次浏览 • 2018-04-02 14:52 • 来自相关话题

运行python requests/urllib2/urllib3 需要sudo/root权限,为什么?

回复

python李魔佛 回复了问题 • 1 人关注 • 1 个回复 • 5194 次浏览 • 2018-01-10 23:36 • 来自相关话题

dataframe重新设置index

回复

python李魔佛 回复了问题 • 1 人关注 • 1 个回复 • 8319 次浏览 • 2017-05-09 23:05 • 来自相关话题

This probably means that Tcl wasn't installed properly [matplotlib][win7]

回复

python李魔佛 发起了问题 • 1 人关注 • 0 个回复 • 8106 次浏览 • 2017-05-05 17:25 • 来自相关话题

在学习装饰器的过程中遇到的奇怪的输出

回复

python李魔佛 发起了问题 • 1 人关注 • 0 个回复 • 3362 次浏览 • 2017-02-09 18:56 • 来自相关话题

pyautogui 在Windows下遇到 WindowsError: [Error 5] Access is denied. 错误

回复

python李魔佛 发起了问题 • 1 人关注 • 0 个回复 • 4455 次浏览 • 2017-01-16 02:03 • 来自相关话题

使用requests 访问https的网页 返回错误: InsecurePlatformWarning: A true SSLContext object is not available

回复

python李魔佛 回复了问题 • 1 人关注 • 1 个回复 • 8020 次浏览 • 2016-08-13 22:52 • 来自相关话题

datetime weekday (可以返回某天是一个星期的第几天)的源码只有return 0

回复

python李魔佛 回复了问题 • 1 人关注 • 1 个回复 • 4687 次浏览 • 2016-08-07 17:57 • 来自相关话题

AttributeError: 'module' object has no attribute 'pyplot'

回复

python李魔佛 回复了问题 • 1 人关注 • 1 个回复 • 7909 次浏览 • 2016-07-28 12:31 • 来自相关话题

ubuntu的pycharm中文注释显示乱码 ?

回复

python李魔佛 回复了问题 • 1 人关注 • 1 个回复 • 8957 次浏览 • 2016-07-25 12:22 • 来自相关话题

pycharm 添加了中文注释后无法运行?

回复

python李魔佛 回复了问题 • 1 人关注 • 1 个回复 • 4672 次浏览 • 2016-07-14 17:56 • 来自相关话题

为什么beautifulsoup的children不能用列表索引index去返回值 ?

回复

python李魔佛 回复了问题 • 1 人关注 • 1 个回复 • 4900 次浏览 • 2016-06-29 22:10 • 来自相关话题

python目录递归?

回复

python李魔佛 回复了问题 • 1 人关注 • 1 个回复 • 4265 次浏览 • 2016-06-07 17:14 • 来自相关话题

B站批量下载某个UP主的所有视频

python李魔佛 发表了文章 • 0 个评论 • 247 次浏览 • 2022-05-21 18:48 • 来自相关话题

B站上不少优秀的学习资源,下载到本地观看,便于快进,多倍速。 也可以放到平板,手机,在没有网络,或者网络条件不佳的环境下观看。
 

 
使用python实现
https://github.com/Rockyzsu/bilibili
 
B站视频下载
自动批量下载B站一个系列的视频

下载某个UP主的所有视频

使用:

下载you-get库,git clone https://github.com/soimort/you-get.git 复制其本地路径,比如/root/you-get/you-get

初次运行,删除history.db 文件, 修改配置文件config.py

START=1 # 下载系列视频的 第一个
END=1 # 下载系列视频的最后一个 , 比如一个系列教程有30个视频, start=5 ,end = 20 下载从第5个到第20个
ID='BV1oK411L7au' # 视频的ID
YOU_GET_PATH='/home/xda/othergit/you-get/you-get' # 你的you-get路径
MINS=1 # 每次循环等待1分钟
user_id = '518973111' # UP主的ID
total_page = 3 # up主的视频的页数
执行 python downloader.py ,进行下载循环

python people.py ,把某个up主的视频链接加入到待下载队列

python add_data.py --id=BV1oK411L7au --start=4 --end=8 下载视频id为BV1oK411L7au的系列教程,从第4开始,到第8个结束,如果只有一个的话,start和end设为1即可。

可以不断地往队列里面添加下载链接。
主要代码:
# @Time : 2019/1/28 14:19
# @File : youtube_downloader.py

import logging
import os
import subprocess
import datetime
import sqlite3
import time
from config import YOU_GET_PATH,MINS
CMD = 'python {} {}'
filename = 'url.txt'


class SQLite():
def __init__(self):
self.conn = sqlite3.connect('history.db')
self.cursor = self.conn.cursor()
self.create_table()

def create_table(self):
create_sql = 'create table if not exists tb_download (url varchar(100),status tinyint,crawltime datetime)'
create_record_tb = 'create table if not exists tb_record (idx varchar(100) PRIMARY KEY,start tinyint,end tinyint,status tinyint)'
self.cursor.execute(create_record_tb)
self.conn.commit()
self.cursor.execute(create_sql)
self.conn.commit()

def exists(self,url):
querySet = 'select * from tb_download where url = ? and status = 1'
self.cursor.execute(querySet,(url,))
ret = self.cursor.fetchone()
return True if ret else False

def insert_history(self,url,status):
query = 'select * from tb_download where url=?'
self.cursor.execute(query,(url,))
ret = self.cursor.fetchone()
current = datetime.datetime.now()

if ret:
insert_sql='update tb_download set status=?,crawltime=? where url = ?'
args=(status,status,current,url)
else:
insert_sql = 'insert into tb_download values(?,?,?)'
args=(url,status,current)

try:
self.cursor.execute(insert_sql,args)
except:
self.conn.rollback()
return False
else:
self.conn.commit()
return True

def get(self):
sql = 'select idx,start,end from tb_record where status=0'
self.cursor.execute(sql)
ret= self.cursor.fetchone()
return ret

def set(self,idx):
print('set status =1')
sql='update tb_record set status=1 where idx=?'
self.cursor.execute(sql,(idx,))
self.conn.commit()

def llogger(filename):
logger = logging.getLogger(filename) # 不加名称设置root logger

logger.setLevel(logging.DEBUG) # 设置输出级别

formatter = logging.Formatter(
'[%(asctime)s][%(filename)s][line: %(lineno)d]\[%(levelname)s] ## %(message)s)',
datefmt='%Y-%m-%d %H:%M:%S')

# 使用FileHandler输出到文件
prefix = os.path.splitext(filename)[0]
fh = logging.FileHandler(prefix + '.log')
fh.setLevel(logging.DEBUG)
fh.setFormatter(formatter)

# 使用StreamHandler输出到屏幕
ch = logging.StreamHandler()
ch.setLevel(logging.DEBUG)
ch.setFormatter(formatter)

# 添加两个Handler
logger.addHandler(ch)
logger.addHandler(fh)
return logger


logger = llogger('download.log')
sql_obj = SQLite()

def run():
while 1:
result = sql_obj.get()
print(result)
if result:
idx=result[0]
start=result[1]
end=result[2]
try:
download_bilibili(idx,start,end)
except:
pass
else:
sql_obj.set(idx)
else:
time.sleep(MINS*60)

def download_bilibili(id,start_page,total_page):
global doc

bilibili_url = 'https://www.bilibili.com/video/{}?p={}'
for i in range(start_page, total_page+1):

next_url = bilibili_url.format(id, i)
if sql_obj.exists(next_url):
print('have download')
continue

try:
command = CMD.format(YOU_GET_PATH, next_url)
p = subprocess.Popen(command, stdout=subprocess.PIPE, stderr=subprocess.PIPE,
shell=True)

output, error = p.communicate()

except Exception as e:
print('has execption')
sql_obj.insert_history(next_url,status=0)
logger.error(e)
continue
else:
output_str = output.decode()
if len(output_str) == 0:
sql_obj.insert_history(next_url,status=0)
logger.info('下载失败')
continue

logger.info('{} has been downloaded !'.format(next_url))
sql_obj.insert_history(next_url,status=1)

run()

  查看全部
B站上不少优秀的学习资源,下载到本地观看,便于快进,多倍速。 也可以放到平板,手机,在没有网络,或者网络条件不佳的环境下观看。
 

 
使用python实现
https://github.com/Rockyzsu/bilibili
 
B站视频下载
自动批量下载B站一个系列的视频

下载某个UP主的所有视频

使用:

下载you-get库,git clone https://github.com/soimort/you-get.git 复制其本地路径,比如/root/you-get/you-get

初次运行,删除history.db 文件, 修改配置文件config.py

START=1 # 下载系列视频的 第一个
END=1 # 下载系列视频的最后一个 , 比如一个系列教程有30个视频, start=5 ,end = 20 下载从第5个到第20个
ID='BV1oK411L7au' # 视频的ID
YOU_GET_PATH='/home/xda/othergit/you-get/you-get' # 你的you-get路径
MINS=1 # 每次循环等待1分钟
user_id = '518973111' # UP主的ID
total_page = 3 # up主的视频的页数
执行 python downloader.py ,进行下载循环

python people.py ,把某个up主的视频链接加入到待下载队列

python add_data.py --id=BV1oK411L7au --start=4 --end=8 下载视频id为BV1oK411L7au的系列教程,从第4开始,到第8个结束,如果只有一个的话,start和end设为1即可。

可以不断地往队列里面添加下载链接。

主要代码:
# @Time : 2019/1/28 14:19
# @File : youtube_downloader.py

import logging
import os
import subprocess
import datetime
import sqlite3
import time
from config import YOU_GET_PATH,MINS
CMD = 'python {} {}'
filename = 'url.txt'


class SQLite():
def __init__(self):
self.conn = sqlite3.connect('history.db')
self.cursor = self.conn.cursor()
self.create_table()

def create_table(self):
create_sql = 'create table if not exists tb_download (url varchar(100),status tinyint,crawltime datetime)'
create_record_tb = 'create table if not exists tb_record (idx varchar(100) PRIMARY KEY,start tinyint,end tinyint,status tinyint)'
self.cursor.execute(create_record_tb)
self.conn.commit()
self.cursor.execute(create_sql)
self.conn.commit()

def exists(self,url):
querySet = 'select * from tb_download where url = ? and status = 1'
self.cursor.execute(querySet,(url,))
ret = self.cursor.fetchone()
return True if ret else False

def insert_history(self,url,status):
query = 'select * from tb_download where url=?'
self.cursor.execute(query,(url,))
ret = self.cursor.fetchone()
current = datetime.datetime.now()

if ret:
insert_sql='update tb_download set status=?,crawltime=? where url = ?'
args=(status,status,current,url)
else:
insert_sql = 'insert into tb_download values(?,?,?)'
args=(url,status,current)

try:
self.cursor.execute(insert_sql,args)
except:
self.conn.rollback()
return False
else:
self.conn.commit()
return True

def get(self):
sql = 'select idx,start,end from tb_record where status=0'
self.cursor.execute(sql)
ret= self.cursor.fetchone()
return ret

def set(self,idx):
print('set status =1')
sql='update tb_record set status=1 where idx=?'
self.cursor.execute(sql,(idx,))
self.conn.commit()

def llogger(filename):
logger = logging.getLogger(filename) # 不加名称设置root logger

logger.setLevel(logging.DEBUG) # 设置输出级别

formatter = logging.Formatter(
'[%(asctime)s][%(filename)s][line: %(lineno)d]\[%(levelname)s] ## %(message)s)',
datefmt='%Y-%m-%d %H:%M:%S')

# 使用FileHandler输出到文件
prefix = os.path.splitext(filename)[0]
fh = logging.FileHandler(prefix + '.log')
fh.setLevel(logging.DEBUG)
fh.setFormatter(formatter)

# 使用StreamHandler输出到屏幕
ch = logging.StreamHandler()
ch.setLevel(logging.DEBUG)
ch.setFormatter(formatter)

# 添加两个Handler
logger.addHandler(ch)
logger.addHandler(fh)
return logger


logger = llogger('download.log')
sql_obj = SQLite()

def run():
while 1:
result = sql_obj.get()
print(result)
if result:
idx=result[0]
start=result[1]
end=result[2]
try:
download_bilibili(idx,start,end)
except:
pass
else:
sql_obj.set(idx)
else:
time.sleep(MINS*60)

def download_bilibili(id,start_page,total_page):
global doc

bilibili_url = 'https://www.bilibili.com/video/{}?p={}'
for i in range(start_page, total_page+1):

next_url = bilibili_url.format(id, i)
if sql_obj.exists(next_url):
print('have download')
continue

try:
command = CMD.format(YOU_GET_PATH, next_url)
p = subprocess.Popen(command, stdout=subprocess.PIPE, stderr=subprocess.PIPE,
shell=True)

output, error = p.communicate()

except Exception as e:
print('has execption')
sql_obj.insert_history(next_url,status=0)
logger.error(e)
continue
else:
output_str = output.decode()
if len(output_str) == 0:
sql_obj.insert_history(next_url,status=0)
logger.info('下载失败')
continue

logger.info('{} has been downloaded !'.format(next_url))
sql_obj.insert_history(next_url,status=1)

run()

 

python3的map是迭代器,不用for循环或者next触发是不会执行的

python李魔佛 发表了文章 • 0 个评论 • 167 次浏览 • 2022-05-21 17:59 • 来自相关话题

最近刚好有位群友咨询,他写的代码如下:
def update_data(id,start,end):
status=0
conn = sqlite3.connect('history.db')
cursor = conn.cursor()
insert_sql ='insert into tb_record values(?,?,?,?)'

try:
cursor.execute(insert_sql,(id,start,end,status))
except Exception as e:
print(e)
print('Error')
else:
conn.commit()
print("successfully insert") 
bv_list = []
for i in range(1, total_page + 1):
bv_list.extend(visit(i))
print(bv_list)
map(lambda x:update_data(x,1,1),bv_list)
作用很简单,就是拿到列表后用map放入到sqlite里面。
但是上面的代码并不起作用。
因为map只是定义了一个迭代器,并没有被触发。
 
可以加一个list(map(lambda x:update_data(x,1,1),bv_list))
这样就可以执行了。 查看全部
最近刚好有位群友咨询,他写的代码如下:
def update_data(id,start,end):
status=0
conn = sqlite3.connect('history.db')
cursor = conn.cursor()
insert_sql ='insert into tb_record values(?,?,?,?)'

try:
cursor.execute(insert_sql,(id,start,end,status))
except Exception as e:
print(e)
print('Error')
else:
conn.commit()
print("successfully insert")
 
bv_list = []
for i in range(1, total_page + 1):
bv_list.extend(visit(i))
print(bv_list)
map(lambda x:update_data(x,1,1),bv_list)

作用很简单,就是拿到列表后用map放入到sqlite里面。
但是上面的代码并不起作用。
因为map只是定义了一个迭代器,并没有被触发。
 
可以加一个list(map(lambda x:update_data(x,1,1),bv_list))
这样就可以执行了。

dataframe如何 遍历所有的列?

python李魔佛 发表了文章 • 0 个评论 • 290 次浏览 • 2022-05-21 02:16 • 来自相关话题

如果遍历行,我们经常会使用df.iterrows(), 而列呢?
可以使用df.items()
 Python pandas.DataFrame.items用法及代码示例
用法:
DataFrame.items()
迭代(列名,系列)对。

遍历 DataFrame 列,返回一个包含列名和内容的元组作为一个系列。

生成(Yield):
label:对象
被迭代的 DataFrame 的列名。

content:Series
属于每个标签的列条目,作为一个系列。

例子:
>>> df = pd.DataFrame({'species':['bear', 'bear', 'marsupial'],
... 'population':[1864, 22000, 80000]},
... index=['panda', 'polar', 'koala'])
>>> df
species population
panda bear 1864
polar bear 22000
koala marsupial 80000
>>> for label, content in df.items():
... print(f'label:{label}')
... print(f'content:{content}', sep='\n')
...
label:species
content:
panda bear
polar bear
koala marsupial
Name:species, dtype:object
label:population
content:
panda 1864
polar 22000
koala 80000
Name:population, dtype:int64 查看全部
如果遍历行,我们经常会使用df.iterrows(), 而列呢?
可以使用df.items()
 
Python pandas.DataFrame.items用法及代码示例
用法:
DataFrame.items()
迭代(列名,系列)对。

遍历 DataFrame 列,返回一个包含列名和内容的元组作为一个系列。

生成(Yield):
label:对象
被迭代的 DataFrame 的列名。

content:Series
属于每个标签的列条目,作为一个系列。

例子:
>>> df = pd.DataFrame({'species':['bear', 'bear', 'marsupial'],
... 'population':[1864, 22000, 80000]},
... index=['panda', 'polar', 'koala'])
>>> df
species population
panda bear 1864
polar bear 22000
koala marsupial 80000
>>> for label, content in df.items():
... print(f'label:{label}')
... print(f'content:{content}', sep='\n')
...
label:species
content:
panda bear
polar bear
koala marsupial
Name:species, dtype:object
label:population
content:
panda 1864
polar 22000
koala 80000
Name:population, dtype:int64

python3 安装demjson 报错 use_2to3 is invalid

python李魔佛 发表了文章 • 0 个评论 • 541 次浏览 • 2022-04-18 20:19 • 来自相关话题

ooking in indexes: https://pypi.douban.com/simple
Collecting demjson==2.2.4
Downloading https://pypi.doubanio.com/pack ... ar.gz (131 kB)
|████████████████████████████████| 131 kB 985 kB/s
ERROR: Command errored out with exit status 1:
command: /root/miniconda3/envs/py37/bin/python -c 'import io, os, sys, setuptools, tokenize; sys.argv[0] = '"'"'/tmp/pip-install-pj0ajcpl/demjson_86a1bab6643c4ed7a7b0c6bb6d3a43a2/setup.py'"'"'; __file__='"'"'/tmp/pip-install-pj0ajcpl/demjson_86a1bab6643c4ed7a7b0c6bb6d3a43a2/setup.py'"'"';f = getattr(tokenize, '"'"'open'"'"', open)(__file__) if os.path.exists(__file__) else io.StringIO('"'"'from setuptools import setup; setup()'"'"');code = f.read().replace('"'"'\r\n'"'"', '"'"'\n'"'"');f.close();exec(compile(code, __file__, '"'"'exec'"'"'))' egg_info --egg-base /tmp/pip-pip-egg-info-7ve4tu87
cwd: /tmp/pip-install-pj0ajcpl/demjson_86a1bab6643c4ed7a7b0c6bb6d3a43a2/
Complete output (3 lines):
/root/miniconda3/envs/py37/lib/python3.7/site-packages/setuptools/dist.py:760: UserWarning: Usage of dash-separated 'index-url' will not be supported in future versions. Please use the underscore name 'index_url' instead
% (opt, underscore_opt)
error in demjson setup command: use_2to3 is invalid.

 记录一下解决办法:
setuptools 降级:
 
pip install --upgrade setuptools==57.5.0
 
然后再pip install demjson 即可
 
如果担心setuptools 修改到系统的其他库,可以创建一个虚拟环境。
然后在虚拟环境里面对setuptools 降级,再安装demjson 
 
来个养眼图: 查看全部
ooking in indexes: https://pypi.douban.com/simple
Collecting demjson==2.2.4
Downloading https://pypi.doubanio.com/pack ... ar.gz (131 kB)
|████████████████████████████████| 131 kB 985 kB/s
ERROR: Command errored out with exit status 1:
command: /root/miniconda3/envs/py37/bin/python -c 'import io, os, sys, setuptools, tokenize; sys.argv[0] = '"'"'/tmp/pip-install-pj0ajcpl/demjson_86a1bab6643c4ed7a7b0c6bb6d3a43a2/setup.py'"'"'; __file__='"'"'/tmp/pip-install-pj0ajcpl/demjson_86a1bab6643c4ed7a7b0c6bb6d3a43a2/setup.py'"'"';f = getattr(tokenize, '"'"'open'"'"', open)(__file__) if os.path.exists(__file__) else io.StringIO('"'"'from setuptools import setup; setup()'"'"');code = f.read().replace('"'"'\r\n'"'"', '"'"'\n'"'"');f.close();exec(compile(code, __file__, '"'"'exec'"'"'))' egg_info --egg-base /tmp/pip-pip-egg-info-7ve4tu87
cwd: /tmp/pip-install-pj0ajcpl/demjson_86a1bab6643c4ed7a7b0c6bb6d3a43a2/
Complete output (3 lines):
/root/miniconda3/envs/py37/lib/python3.7/site-packages/setuptools/dist.py:760: UserWarning: Usage of dash-separated 'index-url' will not be supported in future versions. Please use the underscore name 'index_url' instead
% (opt, underscore_opt)
error in demjson setup command: use_2to3 is invalid.

 记录一下解决办法:
setuptools 降级:
 
pip install --upgrade setuptools==57.5.0
 
然后再pip install demjson 即可
 
如果担心setuptools 修改到系统的其他库,可以创建一个虚拟环境。
然后在虚拟环境里面对setuptools 降级,再安装demjson 
 
来个养眼图:

centos/ubuntu linux 下 python matplotlib 绘图中文标题乱码 解决

Linux李魔佛 发表了文章 • 0 个评论 • 290 次浏览 • 2022-04-15 22:23 • 来自相关话题

出现下图的错误:





 测试代码:import matplotlib as mpl
import matplotlib.pyplot as plt
from matplotlib import rcParams


plt.rcParams['font.family'] = 'sans-serif'
plt.rcParams['font.sans-serif'] = ['simhei'] # 指定默认字体
plt.rcParams['axes.unicode_minus'] = False

plt.title(r'宋体 Times New Roman')
plt.axis('off')
plt.savefig('usestix.png') 
每次换机器都遇到这个问题,所以记录下来,以免日后又在折腾
首先找到一个ttf格式的中文字体,windows系统的可以在C盘的system里面找找。
我用的是微软雅黑。msyh.ttf , 需要的可以到公众号里面取。 后台回复: msyh.ttf 即可
或者simhei.ttfcd /usr/share/fonts/
sudo mkdir chinese
sudo cp simhei.ttf /usr/share/fonts/chinese/
sudo chmod -R 755 /usr/share/fonts/chinese
更新缓存mkfontscale
mkfontdir
fc-cache


注意,如果你的系统没有上述命令,需要使用yum install mkfontscale mkfontdir c-cache安装
 
然后接下来这一步,很重要。
 
清理缓存,不然不会生效!!!!rm -rf ~/.cache/matplotlib


 
matplotlib:
同时把这个msyh.ttf的文件复制到matplotlib库下的文件夹:
~python3.9/site-packages/matplotlib/mpl-data/fonts/ttf/ 
下面
 
然后运行程序,搞定~





 
公众号:可转债量化分析

后台回复 : msyh.ttf 或者 simhei.ttf 即可获取字体文件。 查看全部
出现下图的错误:

20220415_688533_上声电子_zdt.png

 测试代码:
import matplotlib as mpl
import matplotlib.pyplot as plt
from matplotlib import rcParams


plt.rcParams['font.family'] = 'sans-serif'
plt.rcParams['font.sans-serif'] = ['simhei'] # 指定默认字体
plt.rcParams['axes.unicode_minus'] = False

plt.title(r'宋体 Times New Roman')
plt.axis('off')
plt.savefig('usestix.png')
 
每次换机器都遇到这个问题,所以记录下来,以免日后又在折腾
首先找到一个ttf格式的中文字体,windows系统的可以在C盘的system里面找找。
我用的是微软雅黑。msyh.ttf , 需要的可以到公众号里面取。 后台回复: msyh.ttf 即可
或者simhei.ttf
cd /usr/share/fonts/
sudo mkdir chinese
sudo cp simhei.ttf /usr/share/fonts/chinese/
sudo chmod -R 755 /usr/share/fonts/chinese

更新缓存
mkfontscale
mkfontdir
fc-cache


注意,如果你的系统没有上述命令,需要使用yum install mkfontscale mkfontdir c-cache安装
 
然后接下来这一步,很重要。
 
清理缓存,不然不会生效!!!!
rm -rf ~/.cache/matplotlib


 
matplotlib:
同时把这个msyh.ttf的文件复制到matplotlib库下的文件夹:
~python3.9/site-packages/matplotlib/mpl-data/fonts/ttf/ 
下面
 
然后运行程序,搞定~

20220415_688533_上声电子_zdt_(2).png

 
公众号:可转债量化分析

后台回复 : msyh.ttf 或者 simhei.ttf 即可获取字体文件。

python AES 加密 windows和linux平台的不同

python李魔佛 发表了文章 • 0 个评论 • 350 次浏览 • 2022-03-19 11:18 • 来自相关话题

同样一段AES加密的代码,放到了ubuntu可以正常使用,而在windows却报错。
实际两个平台使用pip install安装的aes库不一样。
 
windows报错
File "C:\anaconda\lib\site-packages\Crypto\Cipher\__init__.py", line 77, in _create_cipher
raise TypeError("IV is not meaningful for the ECB mode")
TypeError: IV is not meaningful for the ECB mode

只需要把AES.new() 的参数里面的iv给去掉就可以了。
 

  查看全部
同样一段AES加密的代码,放到了ubuntu可以正常使用,而在windows却报错。
实际两个平台使用pip install安装的aes库不一样。
 
windows报错
  File "C:\anaconda\lib\site-packages\Crypto\Cipher\__init__.py", line 77, in _create_cipher
raise TypeError("IV is not meaningful for the ECB mode")
TypeError: IV is not meaningful for the ECB mode

只需要把AES.new() 的参数里面的iv给去掉就可以了。
 

 

不是所有的bytes都可以转换为string

python李魔佛 发表了文章 • 0 个评论 • 494 次浏览 • 2022-01-14 14:56 • 来自相关话题

byte转为string
b.decode('utf8')
 
如果报错:UnicodeDecodeError: 'utf-8' codec can't decode byte 0xe4 in position 1: invalid continuation byte
说明字节无法字节转为string, 





 
上面的字节是可以正常decode为utf8
 
而改下字节数据





 
所以你试下decode下面的字节:c=b'\x1e\xe4\xd5\x97\x9a#\x99kC\xadD\x7f\x9a\xc2G\x92'
是无法解析的。
 
这个是没有办法的,如果要硬刚 。
可以加入参数errors = ‘replace’
 b.decode('utf8',errors='replace')
这样就不会报错。但是这也只是输出的乱码。
 
正确的姿势是要看看你的字节的最原始编码格式。如果是gbk,那么就应该使用b.decode('gbk')
 
  查看全部
byte转为string
b.decode('utf8')
 
如果报错:
UnicodeDecodeError: 'utf-8' codec can't decode byte 0xe4 in position 1: invalid continuation byte

说明字节无法字节转为string, 

Shutter_2022-01-14-14:49:54.png

 
上面的字节是可以正常decode为utf8
 
而改下字节数据

Shutter_2022-01-14-14:53:40.png

 
所以你试下decode下面的字节:
c=b'\x1e\xe4\xd5\x97\x9a#\x99kC\xadD\x7f\x9a\xc2G\x92'

是无法解析的。
 
这个是没有办法的,如果要硬刚 。
可以加入参数errors = ‘replace’
 
b.decode('utf8',errors='replace')

这样就不会报错。但是这也只是输出的乱码。
 
正确的姿势是要看看你的字节的最原始编码格式。如果是gbk,那么就应该使用b.decode('gbk')
 
 

目前能用的 可以下载youtube 油管视频的方法 合集 亲测

网络李魔佛 发表了文章 • 0 个评论 • 1768 次浏览 • 2021-11-08 13:03 • 来自相关话题

最近的saveform网站下架了油管视频下载的功能。
 
Subject: Discontinuation of Service in the United States

April 16, 2020

Dear Savefrom User:

As you may have heard, our industry has been under strenuous attacks by certain US copyright holders. Because of these attacks, it has become financially impractical for Savefrom to continue to provide services in the United States.

Accordingly, Savefrom will be terminating its services in the United States as of April 28, 2020.

We thank you for your past loyalty and patronage and wish you health and safety during the present health crisis and beyond.

Very truly yours,

Savefrom所以这个网页下载的方法失效了。
 
还有其他 可以使用的方法:
 
方法一   you-get
这是一个python库。
在github下载下来
默认github下载是很卡,且断断续续,具体原因,你懂的。
所以笔者提供一个镜像下载,代码也是最新的。
git clone https://hub.fastgit.org/soimort/you-get.git
运行上面命令即可。
 
然后下载过程就是切换的CMD命令行:
python you-get --debug https://www.youtube.com/watch?v=59a-XNPngrw
 
静静等待即可下载。
 

  查看全部
最近的saveform网站下架了油管视频下载的功能。
 
Subject: Discontinuation of Service in the United States

April 16, 2020

Dear Savefrom User:

As you may have heard, our industry has been under strenuous attacks by certain US copyright holders. Because of these attacks, it has become financially impractical for Savefrom to continue to provide services in the United States.

Accordingly, Savefrom will be terminating its services in the United States as of April 28, 2020.

We thank you for your past loyalty and patronage and wish you health and safety during the present health crisis and beyond.

Very truly yours,

Savefrom
所以这个网页下载的方法失效了。
 
还有其他 可以使用的方法:
 
方法一   you-get
这是一个python库。
在github下载下来
默认github下载是很卡,且断断续续,具体原因,你懂的。
所以笔者提供一个镜像下载,代码也是最新的。
git clone https://hub.fastgit.org/soimort/you-get.git
运行上面命令即可。
 
然后下载过程就是切换的CMD命令行:
python you-get --debug https://www.youtube.com/watch?v=59a-XNPngrw
 
静静等待即可下载。
 

 

通达信公式转python代码

股票量化大师 发表了文章 • 0 个评论 • 1246 次浏览 • 2021-08-12 18:10 • 来自相关话题

本人目前正在做这个转换工作。并定期更新发布在这里。有兴趣的可以参与。
本人目前正在做这个转换工作。并定期更新发布在这里。有兴趣的可以参与。

python 上传文件夹内图片到七牛,同时加入批量删除,单个删除

python李魔佛 发表了文章 • 0 个评论 • 756 次浏览 • 2021-08-07 17:08 • 来自相关话题

先注册好七牛的账户,那都AK和SK两个key
 
然后把key写入到环境变量或者写到下面的python文件里面from qiniu import Auth, put_file,BucketManager,build_batch_delete
import os
import fire

access_key = os.getenv('qiniu_access_key')
secret_key = os.getenv('qiniu_secret_key')

bucket_name = '' # 你的空间名称

HOST ='[url]http://xximg.xxx.com/{}'[/url] # 可以不用填

TEMPLATE = '\n![{}]({})\n\n\n'

def upload(file,category=''):

    #构建鉴权对象
    q = Auth(access_key, secret_key)
    #要上传的空间

    #上传后保存的文件名
    key = category +'/' + os.path.split(file)[1]
    #生成上传 Token,可以指定过期时间等
    token = q.upload_token(bucket_name, key) # 永不过期

    #要上传文件的本地路径
    ret, info = put_file(token, key, file, version='v1') 
    print(ret)
    print(info)
    return HOST.format(ret['key'])

def bulk_upload(path,category=''):
    with open('qiniu_image.md','a+') as fp:
        for file in os.listdir(path):
            full_path = os.path.join(path,file)
            if os.path.isfile(full_path):

                host_url = upload(full_path,category)
                fp.write(TEMPLATE.format(host_url,host_url))


def get_file_info(prefix,limit = 10):
    q = Auth(access_key, secret_key)
    bucket = BucketManager(q)

    delimiter = None
    marker = None
    ret, eof, info = bucket.list(bucket_name, prefix, marker, limit, delimiter)

    # assert len(ret.get('items')) is not None
    url_list=[]
    for item in ret.get('items',):
        url_list.append(item['key'])
    # print(url_list)
    # print(len(url_list))
    return url_list,len(url_list)

def bulk_delete(prefix,limit=None):
    url_list,lens = get_file_info(prefix,limit=limit)
    q = Auth(access_key, secret_key)
    bucket = BucketManager(q)
    ops = build_batch_delete(bucket_name, url_list)
    ret, info = bucket.batch(ops)
    print(info)
    print(ret)

def delete_one(key):
    q = Auth(access_key, secret_key)
    #初始化BucketManager
    bucket = BucketManager(q)
    #你要测试的空间, 并且这个key在你空间中存在
    # key = 'python-logo.png'
    #删除bucket_name 中的文件 key
    ret, info = bucket.delete(bucket_name, key)
    print(info)
    print(ret)
    # assert ret == {}


def bulk_delete_ones(prefix):
    url_list,lens = get_file_info(prefix,limit=10)
    for url in url_list:
        delete_one(url)
        # print(url)

def main(path,category):

    if os.path.isdir(path):

         bulk_upload(path,category)
    elif os.path.isfile(path):
         upload(path,category)
    else:
         raise ValueError('文件不存在')

    get_file_info()
    bulk_delete('resource')
    bulk_delete_ones('resource')
    delete_one('resource/data_beauty.png')

if __name__ == '__main__':
    fire.Fire(main)



然后运行:
python main.py --path='C:\Photo' --category='person'
运行后会上传到七牛的虚拟目录 person目录下
 
如果要删除,bulk_delete批量删除某个前缀或者文件夹的  查看全部
先注册好七牛的账户,那都AK和SK两个key
 
然后把key写入到环境变量或者写到下面的python文件里面
from qiniu import Auth, put_file,BucketManager,build_batch_delete
import os
import fire

access_key = os.getenv('qiniu_access_key')
secret_key = os.getenv('qiniu_secret_key')

bucket_name = '' # 你的空间名称

HOST ='[url]http://xximg.xxx.com/{}'[/url] # 可以不用填

TEMPLATE = '\n![{}]({})\n\n\n'

def upload(file,category=''):

    #构建鉴权对象
    q = Auth(access_key, secret_key)
    #要上传的空间

    #上传后保存的文件名
    key = category +'/' + os.path.split(file)[1]
    #生成上传 Token,可以指定过期时间等
    token = q.upload_token(bucket_name, key) # 永不过期

    #要上传文件的本地路径
    ret, info = put_file(token, key, file, version='v1') 
    print(ret)
    print(info)
    return HOST.format(ret['key'])

def bulk_upload(path,category=''):
    with open('qiniu_image.md','a+') as fp:
        for file in os.listdir(path):
            full_path = os.path.join(path,file)
            if os.path.isfile(full_path):

                host_url = upload(full_path,category)
                fp.write(TEMPLATE.format(host_url,host_url))


def get_file_info(prefix,limit = 10):
    q = Auth(access_key, secret_key)
    bucket = BucketManager(q)

    delimiter = None
    marker = None
    ret, eof, info = bucket.list(bucket_name, prefix, marker, limit, delimiter)

    # assert len(ret.get('items')) is not None
    url_list=[]
    for item in ret.get('items',):
        url_list.append(item['key'])
    # print(url_list)
    # print(len(url_list))
    return url_list,len(url_list)

def bulk_delete(prefix,limit=None):
    url_list,lens = get_file_info(prefix,limit=limit)
    q = Auth(access_key, secret_key)
    bucket = BucketManager(q)
    ops = build_batch_delete(bucket_name, url_list)
    ret, info = bucket.batch(ops)
    print(info)
    print(ret)

def delete_one(key):
    q = Auth(access_key, secret_key)
    #初始化BucketManager
    bucket = BucketManager(q)
    #你要测试的空间, 并且这个key在你空间中存在
    # key = 'python-logo.png'
    #删除bucket_name 中的文件 key
    ret, info = bucket.delete(bucket_name, key)
    print(info)
    print(ret)
    # assert ret == {}


def bulk_delete_ones(prefix):
    url_list,lens = get_file_info(prefix,limit=10)
    for url in url_list:
        delete_one(url)
        # print(url)

def main(path,category):

    if os.path.isdir(path):

         bulk_upload(path,category)
    elif os.path.isfile(path):
         upload(path,category)
    else:
         raise ValueError('文件不存在')

    get_file_info()
    bulk_delete('resource')
    bulk_delete_ones('resource')
    delete_one('resource/data_beauty.png')

if __name__ == '__main__':
    fire.Fire(main)



然后运行:
python main.py --path='C:\Photo' --category='person'
运行后会上传到七牛的虚拟目录 person目录下
 
如果要删除,bulk_delete批量删除某个前缀或者文件夹的 

开通Ptrade Python量化接口 国金证券/国盛证券

股票绫波丽 发表了文章 • 0 个评论 • 6704 次浏览 • 2021-07-06 08:40 • 来自相关话题

目前不少在A股做量化的大部分使用tushare,优矿,米筐,聚宽等等,无论教程,还是实际操作,基本没见有教怎么用程序下单,实盘交易的。

而退而求其次使用按键精灵,模拟点击交易软件进行点击下单,非常不稳定,无法判断下单后是否成交,也无法实时获取行情数据。如果使用tushare或者新浪接口数据,扫描一次全市场的行情用时很久且不稳定,等扫描结束,再下单,此时价格可能已经是几分钟前的了,且此类接口调用次数多是会被封IP的。

笔者使用的是券商提供的量化软件:Ptrade。是恒生电子研发的提供给机构使用的程序化交易软件。提供策略回测,下单API接口,实时行情获取,并且使用的开发语言python,易于上手。

策略回测与实盘交易




研究页面

研究页面,熟悉python jupyter notebook的朋友对这个界面肯定很熟悉。

研究的页面实际就运行你逐行输出调试程序,了解每个函数的具体使用,或者你策略的中途结果调试。





 
回测策略

实际代码需要在回测策略里面写,写完后确定无误,就可以放在仿真环境下真实运行。如果你运行得到的结果很满意,那么就可以直接部署到实盘服务器上。实盘服务器是在券商那边,不需要个人购买服务器,也不需要本地开着这个Ptrade,就是说不需要在个人电脑上一直开着跑,你的最终代码和程序是在券商服务器上部署与运行,除非有报错异常停止,不然在你不暂停或者停止的前提下,可以一直运行下去。












条件满足后下单

可视化量化

同时也提供一些常见的现成的量化策略,选中后只要鼠标点点点也能够自动化跑这些策略了,当然里面很多参数都可以用鼠标点点点修改。





 
接口文档也非常详细:




 
一些常见策略代码:

集合竞价追涨停策略def initialize(context):
# 初始化此策略
# 设置我们要操作的股票池, 这里我们只操作一支股票
g.security = '600570.SS'
set_universe(g.security)
#每天9:23分运行集合竞价处理函数
run_daily(context, aggregate_auction_func, time='9:23')

def aggregate_auction_func(context):
stock = g.security
#最新价
snapshot = get_snapshot(stock)
price = snapshot[stock]['last_px']
#涨停价
up_limit = snapshot[stock]['up_px']
#如果最新价不小于涨停价,买入
if float(price) >= float(up_limit):
order(g.security, 100, limit_price=up_limit)

def handle_data(context, data):
pass
双均线策略def initialize(context):
# 初始化此策略
# 设置我们要操作的股票池, 这里我们只操作一支股票
g.security = '600570.SS'
set_universe(g.security)
pass

#当五日均线高于十日均线时买入,当五日均线低于十日均线时卖出
def handle_data(context, data):
security = g.security

#得到十日历史价格
df = get_history(10, '1d', 'close', security, fq=None, include=False)

# 得到五日均线价格
ma5 = round(df['close'][-5:].mean(), 3)

# 得到十日均线价格
ma10 = round(df['close'][-10:].mean(), 3)

# 取得昨天收盘价
price = data[security]['close']

# 得到当前资金余额
cash = context.portfolio.cash

# 如果当前有余额,并且五日均线大于十日均线
if ma5 > ma10:
# 用所有 cash 买入股票
order_value(security, cash)
# 记录这次买入
log.info("Buying %s" % (security))

# 如果五日均线小于十日均线,并且目前有头寸
elif ma5 < ma10 and get_position(security).amount > 0:
# 全部卖出
order_target(security, 0)
# 记录这次卖出
log.info("Selling %s" % (security))
 
tick级别均线策略

通俗点就是按照秒级别进行操作。def initialize(context):
# 初始化此策略
# 设置我们要操作的股票池, 这里我们只操作一支股票
g.security = '600570.SS'
set_universe(g.security)
#每3秒运行一次主函数
run_interval(context, func, seconds=3)

#盘前准备历史数据
def before_trading_start(context, data):
history = get_history(10, '1d', 'close', g.security, fq='pre', include=False)
g.close_array = history['close'].values

#当五日均线高于十日均线时买入,当五日均线低于十日均线时卖出
def func(context):

stock = g.security

#获取最新价
snapshot = get_snapshot(stock)
price = snapshot[stock]['last_px']

# 得到五日均线价格
days = 5
ma5 = get_MA_day(stock, days, g.close_array[-4:], price)
# 得到十日均线价格
days = 10
ma10 = get_MA_day(stock, days, g.close_array[-9:], price)

# 得到当前资金余额
cash = context.portfolio.cash

# 如果当前有余额,并且五日均线大于十日均线
if ma5 > ma10:
# 用所有 cash 买入股票
order_value(stock, cash)
# 记录这次买入
log.info("Buying %s" % (stock))

# 如果五日均线小于十日均线,并且目前有头寸
elif ma5 < ma10 and get_position(stock).amount > 0:
# 全部卖出
order_target(stock, 0)
# 记录这次卖出
log.info("Selling %s" % (stock))

#计算实时均线函数
def get_MA_day(stock,days,close_array,current_price):
close_sum = close_array[-(days-1):].sum()
MA = (current_price + close_sum)/days
return MA

def handle_data(context, data):
pass
 
macd策略def f_expma(N,m,EXPMA1,price):
a = m/(N+1)
EXPMA2 = a * price + (1 - a)*EXPMA1
return EXPMA2 #2为后一天值

#定义macd函数,输入平滑系数参数、前一日值,输出当日值
def macd(N1,N2,N3,m,EXPMA12_1,EXPMA26_1,DEA1,price):
EXPMA12_2 = f_expma(N1,m,EXPMA12_1,price)
EXPMA26_2 = f_expma(N2,m,EXPMA26_1,price)
DIF2 = EXPMA12_2 - EXPMA26_2
a = m/(N3+1)
DEA2 = a * DIF2 + (1 - a)*DEA1
BAR2=2*(DIF2-DEA2)
return EXPMA12_2,EXPMA26_2,DIF2,DEA2,BAR2

def initialize(context):
global init_price
init_price = None
# 获取沪深300股票
g.security = get_index_stocks('000300.SS')
#g.security = ['600570.SS']
# 设置我们要操作的股票池, 这里我们只操作一支股票
set_universe(g.security)

def handle_data(context, data):
# 获取历史数据,这里只获取了2天的数据,如果希望最终MACD指标结果更准确最好是获取
# 从股票上市至今的所有历史数据,即增加获取的天数
close_price = get_history(2, '1d', field='close', security_list=g.security)
#如果是停牌不进行计算
for security in g.security:
if data[security].is_open >0:
global init_price,EXPMA12_1,EXPMA26_1,EXPMA12_2,EXPMA26_2,DIF1,DIF2,DEA1,DEA2
if init_price is None:
init_price = close_price[security].mean()#nan和N-1个数,mean为N-1个数的均值
EXPMA12_1 = init_price
EXPMA26_1 = init_price
DIF1 = init_price
DEA1 = init_price
# m用于计算平滑系数a=m/(N+1)
m = 2.0
#设定指数平滑基期数
N1 = 12
N2 = 26
N3 = 9
EXPMA12_2,EXPMA26_2,DIF2,DEA2,BAR2 = macd(N1,N2,N3,m,EXPMA12_1,EXPMA26_1,DEA1,close_price[security][-1])
# 取得当前价格
current_price = data[security].price
# 取得当前的现金
cash = context.portfolio.cash
# DIF、DEA均为正,DIF向上突破DEA,买入信号参考
if DIF2 > 0 and DEA2 > 0 and DIF1 < DEA1 and DIF2 > DEA2:
# 计算可以买多少只股票
number_of_shares = int(cash/current_price)
# 购买量大于0时,下单
if number_of_shares > 0:
# 以市单价买入股票,日回测时即是开盘价
order(security, +number_of_shares)
# 记录这次买入
log.info("Buying %s" % (security))
# DIF、DEA均为负,DIF向下突破DEA,卖出信号参考
elif DIF2 < 0 and DEA2 < 0 and DIF1 > DEA1 and DIF2 < DEA2 and get_position(security).amount > 0:
# 卖出所有股票,使这只股票的最终持有量为0
order_target(security, 0)
# 记录这次卖出
log.info("Selling %s" % (security))
# 将今日的值赋给全局变量作为下一次前一日的值
DEA1 = DEA2
DIF1 = DIF2
EXPMA12_1 = EXPMA12_2
EXPMA26_1 = EXPMA26_2
 
软件与交易接口开通条件:

开通该券商后,存入资金指定资金即可开通。开通后股票交易费率万一

本身券商的交易费率为股票万一,可转债沪百万分之五,深十万分之五,基金万0.5,非常厚道。
 
不太了解量化行业的可以了解下,不少面向机构的量化交易软件的佣金是万2.5的,且开户门槛高,基本是500W以上,比如华泰的matic量化的门槛是1千万元起步。

所以笔者还是很推荐目前该券商的量化交易接口。

需要开通咨询了解的朋友可以扫码联系:


开通券商账户后可以 可以先试用,再考虑是否开通量化接口权限 查看全部
目前不少在A股做量化的大部分使用tushare,优矿,米筐,聚宽等等,无论教程,还是实际操作,基本没见有教怎么用程序下单,实盘交易的。

而退而求其次使用按键精灵,模拟点击交易软件进行点击下单,非常不稳定,无法判断下单后是否成交,也无法实时获取行情数据。如果使用tushare或者新浪接口数据,扫描一次全市场的行情用时很久且不稳定,等扫描结束,再下单,此时价格可能已经是几分钟前的了,且此类接口调用次数多是会被封IP的。

笔者使用的是券商提供的量化软件:Ptrade。是恒生电子研发的提供给机构使用的程序化交易软件。提供策略回测,下单API接口,实时行情获取,并且使用的开发语言python,易于上手。

策略回测与实盘交易
UrD5TUnxZD.png

研究页面

研究页面,熟悉python jupyter notebook的朋友对这个界面肯定很熟悉。

研究的页面实际就运行你逐行输出调试程序,了解每个函数的具体使用,或者你策略的中途结果调试。

Hc8f4UtMfW.png

 
回测策略

实际代码需要在回测策略里面写,写完后确定无误,就可以放在仿真环境下真实运行。如果你运行得到的结果很满意,那么就可以直接部署到实盘服务器上。实盘服务器是在券商那边,不需要个人购买服务器,也不需要本地开着这个Ptrade,就是说不需要在个人电脑上一直开着跑,你的最终代码和程序是在券商服务器上部署与运行,除非有报错异常停止,不然在你不暂停或者停止的前提下,可以一直运行下去。

Pig1iRRQnP.png



25YjBEdOqa.png


条件满足后下单

可视化量化

同时也提供一些常见的现成的量化策略,选中后只要鼠标点点点也能够自动化跑这些策略了,当然里面很多参数都可以用鼠标点点点修改。

kTmn9iOXaS.png

 
接口文档也非常详细:
v4QFDpHNpd.png

 
一些常见策略代码:

集合竞价追涨停策略
def initialize(context):
# 初始化此策略
# 设置我们要操作的股票池, 这里我们只操作一支股票
g.security = '600570.SS'
set_universe(g.security)
#每天9:23分运行集合竞价处理函数
run_daily(context, aggregate_auction_func, time='9:23')

def aggregate_auction_func(context):
stock = g.security
#最新价
snapshot = get_snapshot(stock)
price = snapshot[stock]['last_px']
#涨停价
up_limit = snapshot[stock]['up_px']
#如果最新价不小于涨停价,买入
if float(price) >= float(up_limit):
order(g.security, 100, limit_price=up_limit)

def handle_data(context, data):
pass

双均线策略
def initialize(context):
# 初始化此策略
# 设置我们要操作的股票池, 这里我们只操作一支股票
g.security = '600570.SS'
set_universe(g.security)
pass

#当五日均线高于十日均线时买入,当五日均线低于十日均线时卖出
def handle_data(context, data):
security = g.security

#得到十日历史价格
df = get_history(10, '1d', 'close', security, fq=None, include=False)

# 得到五日均线价格
ma5 = round(df['close'][-5:].mean(), 3)

# 得到十日均线价格
ma10 = round(df['close'][-10:].mean(), 3)

# 取得昨天收盘价
price = data[security]['close']

# 得到当前资金余额
cash = context.portfolio.cash

# 如果当前有余额,并且五日均线大于十日均线
if ma5 > ma10:
# 用所有 cash 买入股票
order_value(security, cash)
# 记录这次买入
log.info("Buying %s" % (security))

# 如果五日均线小于十日均线,并且目前有头寸
elif ma5 < ma10 and get_position(security).amount > 0:
# 全部卖出
order_target(security, 0)
# 记录这次卖出
log.info("Selling %s" % (security))

 
tick级别均线策略

通俗点就是按照秒级别进行操作。
def initialize(context):
# 初始化此策略
# 设置我们要操作的股票池, 这里我们只操作一支股票
g.security = '600570.SS'
set_universe(g.security)
#每3秒运行一次主函数
run_interval(context, func, seconds=3)

#盘前准备历史数据
def before_trading_start(context, data):
history = get_history(10, '1d', 'close', g.security, fq='pre', include=False)
g.close_array = history['close'].values

#当五日均线高于十日均线时买入,当五日均线低于十日均线时卖出
def func(context):

stock = g.security

#获取最新价
snapshot = get_snapshot(stock)
price = snapshot[stock]['last_px']

# 得到五日均线价格
days = 5
ma5 = get_MA_day(stock, days, g.close_array[-4:], price)
# 得到十日均线价格
days = 10
ma10 = get_MA_day(stock, days, g.close_array[-9:], price)

# 得到当前资金余额
cash = context.portfolio.cash

# 如果当前有余额,并且五日均线大于十日均线
if ma5 > ma10:
# 用所有 cash 买入股票
order_value(stock, cash)
# 记录这次买入
log.info("Buying %s" % (stock))

# 如果五日均线小于十日均线,并且目前有头寸
elif ma5 < ma10 and get_position(stock).amount > 0:
# 全部卖出
order_target(stock, 0)
# 记录这次卖出
log.info("Selling %s" % (stock))

#计算实时均线函数
def get_MA_day(stock,days,close_array,current_price):
close_sum = close_array[-(days-1):].sum()
MA = (current_price + close_sum)/days
return MA

def handle_data(context, data):
pass

 
macd策略
def f_expma(N,m,EXPMA1,price):
a = m/(N+1)
EXPMA2 = a * price + (1 - a)*EXPMA1
return EXPMA2 #2为后一天值

#定义macd函数,输入平滑系数参数、前一日值,输出当日值
def macd(N1,N2,N3,m,EXPMA12_1,EXPMA26_1,DEA1,price):
EXPMA12_2 = f_expma(N1,m,EXPMA12_1,price)
EXPMA26_2 = f_expma(N2,m,EXPMA26_1,price)
DIF2 = EXPMA12_2 - EXPMA26_2
a = m/(N3+1)
DEA2 = a * DIF2 + (1 - a)*DEA1
BAR2=2*(DIF2-DEA2)
return EXPMA12_2,EXPMA26_2,DIF2,DEA2,BAR2

def initialize(context):
global init_price
init_price = None
# 获取沪深300股票
g.security = get_index_stocks('000300.SS')
#g.security = ['600570.SS']
# 设置我们要操作的股票池, 这里我们只操作一支股票
set_universe(g.security)

def handle_data(context, data):
# 获取历史数据,这里只获取了2天的数据,如果希望最终MACD指标结果更准确最好是获取
# 从股票上市至今的所有历史数据,即增加获取的天数
close_price = get_history(2, '1d', field='close', security_list=g.security)
#如果是停牌不进行计算
for security in g.security:
if data[security].is_open >0:
global init_price,EXPMA12_1,EXPMA26_1,EXPMA12_2,EXPMA26_2,DIF1,DIF2,DEA1,DEA2
if init_price is None:
init_price = close_price[security].mean()#nan和N-1个数,mean为N-1个数的均值
EXPMA12_1 = init_price
EXPMA26_1 = init_price
DIF1 = init_price
DEA1 = init_price
# m用于计算平滑系数a=m/(N+1)
m = 2.0
#设定指数平滑基期数
N1 = 12
N2 = 26
N3 = 9
EXPMA12_2,EXPMA26_2,DIF2,DEA2,BAR2 = macd(N1,N2,N3,m,EXPMA12_1,EXPMA26_1,DEA1,close_price[security][-1])
# 取得当前价格
current_price = data[security].price
# 取得当前的现金
cash = context.portfolio.cash
# DIF、DEA均为正,DIF向上突破DEA,买入信号参考
if DIF2 > 0 and DEA2 > 0 and DIF1 < DEA1 and DIF2 > DEA2:
# 计算可以买多少只股票
number_of_shares = int(cash/current_price)
# 购买量大于0时,下单
if number_of_shares > 0:
# 以市单价买入股票,日回测时即是开盘价
order(security, +number_of_shares)
# 记录这次买入
log.info("Buying %s" % (security))
# DIF、DEA均为负,DIF向下突破DEA,卖出信号参考
elif DIF2 < 0 and DEA2 < 0 and DIF1 > DEA1 and DIF2 < DEA2 and get_position(security).amount > 0:
# 卖出所有股票,使这只股票的最终持有量为0
order_target(security, 0)
# 记录这次卖出
log.info("Selling %s" % (security))
# 将今日的值赋给全局变量作为下一次前一日的值
DEA1 = DEA2
DIF1 = DIF2
EXPMA12_1 = EXPMA12_2
EXPMA26_1 = EXPMA26_2

 
软件与交易接口开通条件:

开通该券商后,存入资金指定资金即可开通。开通后股票交易费率万一

本身券商的交易费率为股票万一,可转债沪百万分之五,深十万分之五,基金万0.5,非常厚道。
 
不太了解量化行业的可以了解下,不少面向机构的量化交易软件的佣金是万2.5的,且开户门槛高,基本是500W以上,比如华泰的matic量化的门槛是1千万元起步。

所以笔者还是很推荐目前该券商的量化交易接口。

需要开通咨询了解的朋友可以扫码联系:


开通券商账户后可以 可以先试用,再考虑是否开通量化接口权限

量化交易接口python A股

股票绫波丽 发表了文章 • 0 个评论 • 2593 次浏览 • 2021-07-01 11:22 • 来自相关话题

不少在A股做量化的基本千篇一律tushare,优矿,米筐,聚宽等等,无论教程也好。
 
实际操作也好,几乎没见几个人教你怎么用程序下单,实盘交易。


稍微好点的easytrader使用的是模拟点击交易软件进行点击下单,基本无法判断下单后是否成交,也无法实时获取行情数据。别跟我说用tushare或者新浪这些接口数据,除非你做的是低频交易。
 
试试扫描一次全市场的行情看看用时多少?等得到满足条件了,再下单,此时价格可能已经在涨停板上封住了。


笔者使用的是券商提供的量化软件:Ptrade。
 
是恒生电子研发的提供给券商机构使用的程序化交易软件。提供策略回测,自动下单功能,使用的开发语言python。


策略回测与实盘交易






研究页面


研究页面,熟悉python jupyter notebook的朋友对这个界面肯定很熟悉。

研究的页面实际就运行你逐行输出调试程序,了解每个函数的具体使用,或者你策略的中途结果调试。








回测策略

实际代码需要在回测策略里面写,写完后确定无误,就可以放在仿真环境下真实运行。
 
如果你运行得到的结果很满意,那么就可以直接部署到实盘服务器上。
 
实盘服务器是在券商那边,不需要个人购买服务器,也不需要本地开着这个Ptrade,就是说不需要在个人电脑上一直开着跑,你的最终代码和程序是在券商服务器上部署与运行,除非有报错异常停止,不然在你不暂停或者停止的前提下,可以一直运行下去。












条件满足后下单

可视化量化

同时也提供一些常见的现成的量化策略,选中后只要鼠标点点点也能够自动化跑这些策略了,当然里面很多参数都可以用鼠标点点点修改。






接口文档也非常详细:








一些常见策略代码:

双均线策略 def initialize(context):
2 # 初始化此策略
3 # 设置我们要操作的股票池, 这里我们只操作一支股票
4 g.security = '600570.SS'
5 set_universe(g.security)
6 pass
7
8 #当五日均线高于十日均线时买入,当五日均线低于十日均线时卖出
9 def handle_data(context, data):
10 security = g.security
11
12 #得到十日历史价格
13 df = get_history(10, '1d', 'close', security, fq=None, include=False)
14
15 # 得到五日均线价格
16 ma5 = round(df['close'][-5:].mean(), 3)
17
18 # 得到十日均线价格
19 ma10 = round(df['close'][-10:].mean(), 3)
20
21 # 取得昨天收盘价
22 price = data[security]['close']
23
24 # 得到当前资金余额
25 cash = context.portfolio.cash
26
27 # 如果当前有余额,并且五日均线大于十日均线
28 if ma5 > ma10:
29 # 用所有 cash 买入股票
30 order_value(security, cash)
31 # 记录这次买入
32 log.info("Buying %s" % (security))
33
34 # 如果五日均线小于十日均线,并且目前有头寸
35 elif ma5 < ma10 and get_position(security).amount > 0:
36 # 全部卖出
37 order_target(security, 0)
38 # 记录这次卖出
39 log.info("Selling %s" % (security))
 macd策略
指数平滑均线函数,以price计算,可以选择收盘、开盘价等价格,N为时间周期,m用于计算平滑系数a=m/(N+1),EXPMA1为前一日值 def f_expma(N,m,EXPMA1,price):
2 a = m/(N+1)
3 EXPMA2 = a * price + (1 - a)*EXPMA1
4 return EXPMA2 #2为后一天值
5
6 #定义macd函数,输入平滑系数参数、前一日值,输出当日值
7 def macd(N1,N2,N3,m,EXPMA12_1,EXPMA26_1,DEA1,price):
8 EXPMA12_2 = f_expma(N1,m,EXPMA12_1,price)
9 EXPMA26_2 = f_expma(N2,m,EXPMA26_1,price)
10 DIF2 = EXPMA12_2 - EXPMA26_2
11 a = m/(N3+1)
12 DEA2 = a * DIF2 + (1 - a)*DEA1
13 BAR2=2*(DIF2-DEA2)
14 return EXPMA12_2,EXPMA26_2,DIF2,DEA2,BAR2
15
16 def initialize(context):
17 global init_price
18 init_price = None
19 # 获取沪深300股票
20 g.security = get_index_stocks('000300.SS')
21 #g.security = ['600570.SS']
22 # 设置我们要操作的股票池, 这里我们只操作一支股票
23 set_universe(g.security)
24
25 def handle_data(context, data):
26 # 获取历史数据,这里只获取了2天的数据,如果希望最终MACD指标结果更准确最好是获取
27 # 从股票上市至今的所有历史数据,即增加获取的天数
28 close_price = get_history(2, '1d', field='close', security_list=g.security)
29 #如果是停牌不进行计算
30 for security in g.security:
31 if data[security].is_open >0:
32 global init_price,EXPMA12_1,EXPMA26_1,EXPMA12_2,EXPMA26_2,DIF1,DIF2,DEA1,DEA2
33 if init_price is None:
34 init_price = close_price[security].mean()#nan和N-1个数,mean为N-1个数的均值
35 EXPMA12_1 = init_price
36 EXPMA26_1 = init_price
37 DIF1 = init_price
38 DEA1 = init_price
39 # m用于计算平滑系数a=m/(N+1)
40 m = 2.0
41 #设定指数平滑基期数
42 N1 = 12
43 N2 = 26
44 N3 = 9
45 EXPMA12_2,EXPMA26_2,DIF2,DEA2,BAR2 = macd(N1,N2,N3,m,EXPMA12_1,EXPMA26_1,DEA1,close_price[security][-1])
46 # 取得当前价格
47 current_price = data[security].price
48 # 取得当前的现金
49 cash = context.portfolio.cash
50 # DIF、DEA均为正,DIF向上突破DEA,买入信号参考
51 if DIF2 > 0 and DEA2 > 0 and DIF1 < DEA1 and DIF2 > DEA2:
52 # 计算可以买多少只股票
53 number_of_shares = int(cash/current_price)
54 # 购买量大于0时,下单
55 if number_of_shares > 0:
56 # 以市单价买入股票,日回测时即是开盘价
57 order(security, +number_of_shares)
58 # 记录这次买入
59 log.info("Buying %s" % (security))
60 # DIF、DEA均为负,DIF向下突破DEA,卖出信号参考
61 elif DIF2 < 0 and DEA2 < 0 and DIF1 > DEA1 and DIF2 < DEA2 and get_position(security).amount > 0:
62 # 卖出所有股票,使这只股票的最终持有量为0
63 order_target(security, 0)
64 # 记录这次卖出
65 log.info("Selling %s" % (security))
66 # 将今日的值赋给全局变量作为下一次前一日的值
67 DEA1 = DEA2
68 DIF1 = DIF2
69 EXPMA12_1 = EXPMA12_2
70 EXPMA26_1 = EXPMA26_2
 
同时也支持可转债T+0交易,笔者平时也主要是进行的可转债的交易。
 
软件与交易接口开通条件,

开通改券商后,存入资金50W放2周即可开通,开通后可取出。

本身券商的交易费率为股票万一免五,可转债百万分之二,非常厚道,因为不少面向机构的量化交易软件的佣金是万2.5的,笔者之前在国金渠道开的就是,坑的一比。所以笔者还是很推荐目前该券商的量化交易接口。
 
需要可以关注公众号获取开通链接: 查看全部
不少在A股做量化的基本千篇一律tushare,优矿,米筐,聚宽等等,无论教程也好。
 
实际操作也好,几乎没见几个人教你怎么用程序下单,实盘交易。


稍微好点的easytrader使用的是模拟点击交易软件进行点击下单,基本无法判断下单后是否成交,也无法实时获取行情数据。别跟我说用tushare或者新浪这些接口数据,除非你做的是低频交易。
 
试试扫描一次全市场的行情看看用时多少?等得到满足条件了,再下单,此时价格可能已经在涨停板上封住了。


笔者使用的是券商提供的量化软件:Ptrade
 
是恒生电子研发的提供给券商机构使用的程序化交易软件。提供策略回测,自动下单功能,使用的开发语言python。


策略回测与实盘交易

UrD5TUnxZD.png


研究页面


研究页面,熟悉python jupyter notebook的朋友对这个界面肯定很熟悉。

研究的页面实际就运行你逐行输出调试程序,了解每个函数的具体使用,或者你策略的中途结果调试。


Hc8f4UtMfW.png



回测策略

实际代码需要在回测策略里面写,写完后确定无误,就可以放在仿真环境下真实运行。
 
如果你运行得到的结果很满意,那么就可以直接部署到实盘服务器上。
 
实盘服务器是在券商那边,不需要个人购买服务器,也不需要本地开着这个Ptrade,就是说不需要在个人电脑上一直开着跑,你的最终代码和程序是在券商服务器上部署与运行,除非有报错异常停止,不然在你不暂停或者停止的前提下,可以一直运行下去。

Pig1iRRQnP.png



25YjBEdOqa.png


条件满足后下单

可视化量化

同时也提供一些常见的现成的量化策略,选中后只要鼠标点点点也能够自动化跑这些策略了,当然里面很多参数都可以用鼠标点点点修改。

kTmn9iOXaS.png


接口文档也非常详细:


v4QFDpHNpd.png



一些常见策略代码:

双均线策略
    def initialize(context):
2 # 初始化此策略
3 # 设置我们要操作的股票池, 这里我们只操作一支股票
4 g.security = '600570.SS'
5 set_universe(g.security)
6 pass
7
8 #当五日均线高于十日均线时买入,当五日均线低于十日均线时卖出
9 def handle_data(context, data):
10 security = g.security
11
12 #得到十日历史价格
13 df = get_history(10, '1d', 'close', security, fq=None, include=False)
14
15 # 得到五日均线价格
16 ma5 = round(df['close'][-5:].mean(), 3)
17
18 # 得到十日均线价格
19 ma10 = round(df['close'][-10:].mean(), 3)
20
21 # 取得昨天收盘价
22 price = data[security]['close']
23
24 # 得到当前资金余额
25 cash = context.portfolio.cash
26
27 # 如果当前有余额,并且五日均线大于十日均线
28 if ma5 > ma10:
29 # 用所有 cash 买入股票
30 order_value(security, cash)
31 # 记录这次买入
32 log.info("Buying %s" % (security))
33
34 # 如果五日均线小于十日均线,并且目前有头寸
35 elif ma5 < ma10 and get_position(security).amount > 0:
36 # 全部卖出
37 order_target(security, 0)
38 # 记录这次卖出
39 log.info("Selling %s" % (security))

 macd策略
指数平滑均线函数,以price计算,可以选择收盘、开盘价等价格,N为时间周期,m用于计算平滑系数a=m/(N+1),EXPMA1为前一日值
    def f_expma(N,m,EXPMA1,price):
2 a = m/(N+1)
3 EXPMA2 = a * price + (1 - a)*EXPMA1
4 return EXPMA2 #2为后一天值
5
6 #定义macd函数,输入平滑系数参数、前一日值,输出当日值
7 def macd(N1,N2,N3,m,EXPMA12_1,EXPMA26_1,DEA1,price):
8 EXPMA12_2 = f_expma(N1,m,EXPMA12_1,price)
9 EXPMA26_2 = f_expma(N2,m,EXPMA26_1,price)
10 DIF2 = EXPMA12_2 - EXPMA26_2
11 a = m/(N3+1)
12 DEA2 = a * DIF2 + (1 - a)*DEA1
13 BAR2=2*(DIF2-DEA2)
14 return EXPMA12_2,EXPMA26_2,DIF2,DEA2,BAR2
15
16 def initialize(context):
17 global init_price
18 init_price = None
19 # 获取沪深300股票
20 g.security = get_index_stocks('000300.SS')
21 #g.security = ['600570.SS']
22 # 设置我们要操作的股票池, 这里我们只操作一支股票
23 set_universe(g.security)
24
25 def handle_data(context, data):
26 # 获取历史数据,这里只获取了2天的数据,如果希望最终MACD指标结果更准确最好是获取
27 # 从股票上市至今的所有历史数据,即增加获取的天数
28 close_price = get_history(2, '1d', field='close', security_list=g.security)
29 #如果是停牌不进行计算
30 for security in g.security:
31 if data[security].is_open >0:
32 global init_price,EXPMA12_1,EXPMA26_1,EXPMA12_2,EXPMA26_2,DIF1,DIF2,DEA1,DEA2
33 if init_price is None:
34 init_price = close_price[security].mean()#nan和N-1个数,mean为N-1个数的均值
35 EXPMA12_1 = init_price
36 EXPMA26_1 = init_price
37 DIF1 = init_price
38 DEA1 = init_price
39 # m用于计算平滑系数a=m/(N+1)
40 m = 2.0
41 #设定指数平滑基期数
42 N1 = 12
43 N2 = 26
44 N3 = 9
45 EXPMA12_2,EXPMA26_2,DIF2,DEA2,BAR2 = macd(N1,N2,N3,m,EXPMA12_1,EXPMA26_1,DEA1,close_price[security][-1])
46 # 取得当前价格
47 current_price = data[security].price
48 # 取得当前的现金
49 cash = context.portfolio.cash
50 # DIF、DEA均为正,DIF向上突破DEA,买入信号参考
51 if DIF2 > 0 and DEA2 > 0 and DIF1 < DEA1 and DIF2 > DEA2:
52 # 计算可以买多少只股票
53 number_of_shares = int(cash/current_price)
54 # 购买量大于0时,下单
55 if number_of_shares > 0:
56 # 以市单价买入股票,日回测时即是开盘价
57 order(security, +number_of_shares)
58 # 记录这次买入
59 log.info("Buying %s" % (security))
60 # DIF、DEA均为负,DIF向下突破DEA,卖出信号参考
61 elif DIF2 < 0 and DEA2 < 0 and DIF1 > DEA1 and DIF2 < DEA2 and get_position(security).amount > 0:
62 # 卖出所有股票,使这只股票的最终持有量为0
63 order_target(security, 0)
64 # 记录这次卖出
65 log.info("Selling %s" % (security))
66 # 将今日的值赋给全局变量作为下一次前一日的值
67 DEA1 = DEA2
68 DIF1 = DIF2
69 EXPMA12_1 = EXPMA12_2
70 EXPMA26_1 = EXPMA26_2

 
同时也支持可转债T+0交易,笔者平时也主要是进行的可转债的交易。
 
软件与交易接口开通条件,

开通改券商后,存入资金50W放2周即可开通,开通后可取出。

本身券商的交易费率为股票万一免五,可转债百万分之二,非常厚道,因为不少面向机构的量化交易软件的佣金是万2.5的,笔者之前在国金渠道开的就是,坑的一比。所以笔者还是很推荐目前该券商的量化交易接口。
 
需要可以关注公众号获取开通链接:

pyautogui无法再远程桌面最小化或者断线后进行截图

python爬虫李魔佛 发表了文章 • 0 个评论 • 1276 次浏览 • 2021-04-29 17:19 • 来自相关话题

搜索了一圈,似乎无解。
知道的朋友可以私信下我。
 
国外的论坛也找不到答案,只能一直开着屏幕了。。。。
搜索了一圈,似乎无解。
知道的朋友可以私信下我。
 
国外的论坛也找不到答案,只能一直开着屏幕了。。。。

pip install peewee : AttributeError: 'str' object has no attribute 'decode'

python李魔佛 发表了文章 • 0 个评论 • 1083 次浏览 • 2021-04-26 23:58 • 来自相关话题

 
ERROR: Command errored out with exit status 1:
command: 'C:\anaconda\python.exe' -c 'import sys, setuptools, tokenize; sys.argv[0] = '"'"'C:\\Users\\xda\\AppData\
\Local\\Temp\\pip-install-ftotbzih\\peewee\\setup.py'"'"'; __file__='"'"'C:\\Users\\xda\\AppData\\Local\\Temp\\pip-insta
ll-ftotbzih\\peewee\\setup.py'"'"';f=getattr(tokenize, '"'"'open'"'"', open)(__file__);code=f.read().replace('"'"'\r\n'"
'"', '"'"'\n'"'"');f.close();exec(compile(code, __file__, '"'"'exec'"'"'))' egg_info --egg-base 'C:\Users\xda\AppData\Lo
cal\Temp\pip-pip-egg-info-8ou7yi3i'
cwd: C:\Users\xda\AppData\Local\Temp\pip-install-ftotbzih\peewee\
Complete output (15 lines):
Traceback (most recent call last):
File "<string>", line 1, in <module>
File "C:\Users\xda\AppData\Local\Temp\pip-install-ftotbzih\peewee\setup.py", line 99, in <module>
elif not _have_sqlite_extension_support():
File "C:\Users\xda\AppData\Local\Temp\pip-install-ftotbzih\peewee\setup.py", line 76, in _have_sqlite_extension_su
pport
compiler.compile([src_file], output_dir=tmp_dir),
File "C:\anaconda\lib\distutils\_msvccompiler.py", line 327, in compile
self.initialize()
File "C:\anaconda\lib\distutils\_msvccompiler.py", line 224, in initialize
vc_env = _get_vc_env(plat_spec)
File "C:\anaconda\lib\site-packages\setuptools\msvc.py", line 314, in msvc14_get_vc_env
return _msvc14_get_vc_env(plat_spec)
File "C:\anaconda\lib\site-packages\setuptools\msvc.py", line 273, in _msvc14_get_vc_env
out = subprocess.check_output(
AttributeError: 'str' object has no attribute 'decode'
----------------------------------------
ERROR: Command errored out with exit status 1: python setup.py egg_info Check the logs for full command output.
同样的编码问题,同样的解决方法:
找到文件msvc.py
大概在276行:
out = subprocess.check_output(
'cmd /u /c "{}" {} && set'.format(vcvarsall, plat_spec),
stderr=subprocess.STDOUT,
)
# ).decode('utf-16le', errors='replace')把decode的部分注释掉即可
  查看全部
 
    ERROR: Command errored out with exit status 1:
command: 'C:\anaconda\python.exe' -c 'import sys, setuptools, tokenize; sys.argv[0] = '"'"'C:\\Users\\xda\\AppData\
\Local\\Temp\\pip-install-ftotbzih\\peewee\\setup.py'"'"'; __file__='"'"'C:\\Users\\xda\\AppData\\Local\\Temp\\pip-insta
ll-ftotbzih\\peewee\\setup.py'"'"';f=getattr(tokenize, '"'"'open'"'"', open)(__file__);code=f.read().replace('"'"'\r\n'"
'"', '"'"'\n'"'"');f.close();exec(compile(code, __file__, '"'"'exec'"'"'))' egg_info --egg-base 'C:\Users\xda\AppData\Lo
cal\Temp\pip-pip-egg-info-8ou7yi3i'
cwd: C:\Users\xda\AppData\Local\Temp\pip-install-ftotbzih\peewee\
Complete output (15 lines):
Traceback (most recent call last):
File "<string>", line 1, in <module>
File "C:\Users\xda\AppData\Local\Temp\pip-install-ftotbzih\peewee\setup.py", line 99, in <module>
elif not _have_sqlite_extension_support():
File "C:\Users\xda\AppData\Local\Temp\pip-install-ftotbzih\peewee\setup.py", line 76, in _have_sqlite_extension_su
pport
compiler.compile([src_file], output_dir=tmp_dir),
File "C:\anaconda\lib\distutils\_msvccompiler.py", line 327, in compile
self.initialize()
File "C:\anaconda\lib\distutils\_msvccompiler.py", line 224, in initialize
vc_env = _get_vc_env(plat_spec)
File "C:\anaconda\lib\site-packages\setuptools\msvc.py", line 314, in msvc14_get_vc_env
return _msvc14_get_vc_env(plat_spec)
File "C:\anaconda\lib\site-packages\setuptools\msvc.py", line 273, in _msvc14_get_vc_env
out = subprocess.check_output(
AttributeError: 'str' object has no attribute 'decode'
----------------------------------------
ERROR: Command errored out with exit status 1: python setup.py egg_info Check the logs for full command output.

同样的编码问题,同样的解决方法:
找到文件msvc.py
大概在276行:
        out = subprocess.check_output(
'cmd /u /c "{}" {} && set'.format(vcvarsall, plat_spec),
stderr=subprocess.STDOUT,
)
# ).decode('utf-16le', errors='replace')
把decode的部分注释掉即可
 

百度AI的语音识别无法识别到英文原因

深度学习李魔佛 发表了文章 • 0 个评论 • 1245 次浏览 • 2021-04-25 13:23 • 来自相关话题

大概率是因为你的音频码率和要求的不一致造成的。
可以尝试使用ffmepg转下码:
ffmpeg.exe -i 3.mp3 -ac 1 -ar 16000 16k.wav试过后就能够正常识别到语音内容了。
 
大概率是因为你的音频码率和要求的不一致造成的。
可以尝试使用ffmepg转下码:
ffmpeg.exe -i 3.mp3 -ac 1 -ar 16000 16k.wav
试过后就能够正常识别到语音内容了。