文章为《量化交易之路》 阿布@著 -----笔记

python

#命名元组
from collections import namedtuple
# 有序字典
from collections import OrderedDict
# 高阶函数
map()
filter()
reduce()
# 偏函数
from functools import partial
# property装饰器

# 使用itertools可以提升效率
# 顺序组合元素
for item in itertools.permutations([1,2,3])
	print(item)
	
# 不考虑顺序，不回放数据
itertools.combinations()
# 不考虑顺序，有回放数据
itertools.combinations_with_replacement()
# 笛卡尔积
itertools.product()
# 合并但是节省空间
itertools.chain()

# numba模块进行静态编译原始函数
numba.jit()

abupy包

numpy

import numpy as np

np.all() # 是否全都为true
np.any()  # 是否有元素为true
np.maximum()
np.minimum()
np.unique()
np.diff()  # 
np.where()  # 类似三元运算

正态分布图

import scipy.stats as scs
import numpy as np
import matplotlib.pyplot as plt

keep_days=50
# 股票数量
stock_cnt=200
# 交易日数
view_days=504
stock_day_change=np.random.standard_normal((stock_cnt,view_days))

stock_mean=stock_day_change[0].mean()
stock_std=stock_day_change[0].std()
plt.hist(stock_day_change[0],bins=50,density=True)
fit_linspace=np.linspace(stock_day_change[0].min(),stock_day_change[0].max())
pdf=scs.norm(stock_mean,stock_std).pdf(fit_linspace)
plt.plot(fit_linspace,pdf,lw=2,c='r')
plt.show()

正态分布策略

买入跌幅最大的3只个股，进行测试

import numpy as np
import matplotlib.pyplot as plt


# 保留50天数据进行策略验证
keep_days=50
# 股票数量
stock_cnt=200
# 交易日数
view_days=504
stock_day_change=np.random.standard_normal((stock_cnt,view_days))

stock_day_change_test=stock_day_change[:stock_cnt,0:view_days-keep_days]
# 跌幅最大的3只个股
# print(np.sort(np.sum(stock_day_change_test,axis=1))[:3])
# 符合条件的 股票序号
stock_lower_array=np.argsort(np.sum(stock_day_change_test,axis=1))[:3]
print(stock_lower_array)
def show_buy_lower(stock_ind):
    #设置一个一行两列的可视化图表
    _,axs=plt.subplots(nrows=1,ncols=2,figsize=(16,5))
    # 绘制454天走势图 np.cumsum() 序列连续求和
    axs[0].plot(np.arange(0,view_days-keep_days),stock_day_change_test[stock_ind].cumsum())
    cs_buy=stock_day_change[stock_ind][view_days-keep_days:view_days].cumsum()
    axs[1].plot(np.arange(view_days-keep_days,view_days),cs_buy)
    return cs_buy[-1]
# 最后输出的盈亏比例
profit=0
for stock_ind in stock_lower_array:
    profit+=show_buy_lower(stock_ind)
print('买入第{}只股票，从第454个交易日开始持有盈利：{:.2f}'.format(stock_lower_array,profit))
plt.show()

伯努力分布

import numpy as np
import matplotlib.pyplot as plt

# 设置200个赌徒
gamblers=200

def casino(win_rate,win_once=1,loss_once=1,commission=0.01):
    my_money=1000000
    play_cnt=10000000
    commission=commission
    for _ in np.arange(0,play_cnt):
        # 使用伯努力分布，根据win_rate来获取输赢
        w=np.random.binomial(1,win_rate)
        if w:
            my_money +=win_once # 赢了
        else:
            my_money-=loss_once # 输了
        my_money-=commission  # 手续费
        if not my_money:
            break
    return my_money
# 没有 老千 没有手续费
heaven_moneys=[casino(0.5,commission=0) for _ in np.arange(0,gamblers)]
_=plt.hist(heaven_moneys,bins=30)
# 有 老千 没有手续费  胜率只能0.4
cheat_moneys=[casino(0.4,commission=0) for _ in np.arange(0,gamblers)]
_=plt.hist(cheat_moneys,bins=30)
# 有手续费 没有老千
commission_moneys=[casino(0.5,commission=0.01) for _ in np.arange(0,gamblers)]
_=plt.hist(commission_moneys,bins=30)
# 较为真实的模拟
moneys=[casino(0.5,win_once=1.02,loss_once=0.98,commission=0.01,) for _ in np.arange(0,gamblers)]
_=plt.hist(moneys,bins=30)

plt.show()

Pandas

import pandas as pd
pd.resample('21D',how='mean') #重采样 21日均线

df_stock0=df['股票0']
df_stock0.cumsum().plot()  # 画出走势图

df_stock0_5=df_stock0.cumsum().resample('5D',how='ohlc')# 周k
df_stock0_20=df_stock0.cumsum().resample('21D',how='ohlc') # 月k

from abupy import ABuMarketDrawing
# 画出蜡烛图
ABuMarketDrawing.plot_candle_stick(df_stock0_5.index,
								df_stock0_5['open'].values,
								df_stock0_5['high'].values,
								df_stock0_5['low'].values,
								df_stock0_5['close'].values,
								np.random.random(len(df_stock0_5)),
								None,'stock',any_sum=False,html_bk=False,save=False
								)

# 获得数据并画图
import matplotlib.pyplot as plt
from abupy import ABuSymbolPd
tsla_df=ABuSymbolPd.make_kl_df('usTSLA',n_folds=2)
# print(tsla_df.tail())

tsla_df[['close','volume']].plot(subplots=True,style=['r','g'],grid=True)
tsla_df.info()
tsla_df.describe()

tsla_df[np.abs(tsla_df.netChangeRatio)>8]
tsla_df.sort_index(by='netChangeRatio',ascending=False)[:5]  # 涨幅最大的5个交易日
# 删除nan值
tsla_df.dropna()
# 填充函数
tsla_df.fillna()
# 对序列从第二项开始向前做减法后除以前一项，例如涨跌幅序列计算
tsla_df.close.pct_change()[:3]
# 对数据进行保留小数位数操作
np.round()
plt.show()

# 合并
pd.concat()
DataFrame.append() # 
pd.merge()

寻找股票异动涨跌幅阀值

# 涨跌额直方图
import matplotlib.pyplot as plt
import pandas as pd
import numpy as np
from abupy import ABuSymbolPd
tsla_df=ABuSymbolPd.make_kl_df('usTSLA',n_folds=2)
# print(tsla_df.describe())
tsla_df.p_change.hist(bins=80)
cats=pd.cut(np.abs(tsla_df.p_change),10)
cats.value_counts()
plt.show() 

数据离散化

import pandas as pd
import numpy as np
from abupy import ABuSymbolPd
tsla_df=ABuSymbolPd.make_kl_df('usTSLA',n_folds=2)

# 数据离散化
bins=[-np.inf,-7.0,-5,-3,0,3,5,7,np.inf]
cats=pd.cut(tsla_df.p_change,bins=bins)
cats.value_counts()
# 生成离散化的哑变量矩阵
p_change_dummies=pd.get_dummies(cats,prefix='p_change')
print(p_change_dummies.tail())

星期几是这个票的好日子

import matplotlib.pyplot as plt
import pandas as pd
import numpy as np
from abupy import ABuSymbolPd
tsla_df=ABuSymbolPd.make_kl_df('usTSLA',n_folds=2)

tsla_df['positive']=np.where(tsla_df.p_change>0,1,0)
print(tsla_df.tail())
# 构建交叉表
xt=pd.crosstab(tsla_df.date_week,tsla_df.positive)
xt_pct=xt.div(xt.sum(1).astype(float),axis=0)
xt_pct.plot(figsize=(8,5),kind='bar',stacked=True,title='date_week')
plt.xlabel('date_week')
plt.ylabel('positive')
plt.show()
# 构建透视表
ss=tsla_df.pivot_table(['positive'],index=['date_week'])
print(ss)
# 分类统计
dd=tsla_df.groupby(['date_week','positive'])['positive'].count()
print(dd)

跳空缺口

import matplotlib.pyplot as plt
import pandas as pd
import numpy as np
from abupy import ABuSymbolPd
tsla_df=ABuSymbolPd.make_kl_df('usTSLA',n_folds=2)
# print(tsla_df.head())
jump=tsla_df.close.median()*0.03
# print(jump)

jump_pd=pd.DataFrame()

def judge_jump(today):
    global jump_pd
    if today.p_change>0 and (today.low-today.pre_close) > jump:
        # 向上跳空
        today['jump']=1
        today['jump_power'] = (today.low-today.pre_close)/jump
        jump_pd = jump_pd.append(today)
    elif today.p_change<0 and (today.pre_close-today.high)> jump:
        #向下跳空
        today['jump']=-1
        today['jump_power'] = (today.pre_close-today.high) / jump
        jump_pd = jump_pd.append(today)

for kl_index in np.arange(0,tsla_df.shape[0]):
    today = tsla_df.ix[kl_index]
    judge_jump(today)

# # 另一种方式 更优雅
# jump_pd = pd.DataFrame()
# tsla_df.apply(judge_jump, axis=1)


res=jump_pd.filter(['jump','jump_power','close','date','pre_close','p_change'])
print(res)
# 此处图形不能呈现，报错 No module named 'matplotlib.mpl_finance'
from abupy import ABuMarketDrawing
ABuMarketDrawing.plot_candle_form_klpd(tsla_df,view_indexs=jump_pd.index)
plt.show()