Python 量化代碼

1、# coding: utf-8import numpy as npimport pandas as pdimport matplotlib.pyplot as pltfrom _future_ import division# 獲取數(shù)據(jù)函數(shù)def get_stock_data(stock_code, index_code, start_date, end_date): """ :param stock_code: 股票代碼，例如sz000002 :param index_code: 指數(shù)代碼，例如sh000001 :param start_date: 回測開始日期

2、，例如1991-1-30' :param end_date: 回測結(jié)束日期，例如2015-12-31 :return: 函數(shù)返回其他函數(shù)的各參數(shù)序列 """ # 此處為存放csv文件的本地路徑，請自行改正地址.注意windows和mac系統(tǒng),斜杠的方向不一樣 stock_data = pd.read_csv(r'G:財通實習(xí)歷史日線數(shù)據(jù)_樣本%282013 2014年數(shù)據(jù)%292all_trading_datastock datash600000.csv', parse_dates='date') benchmark = p

3、d.read_csv(r'G:財通實習(xí)歷史日線數(shù)據(jù)_樣本%282013 2014年數(shù)據(jù)%292all_trading_dataindex datash000001.csv', parse_dates='date') date = pd.date_range(2016-01-01,2016-03-15) # 生成日期序列 # 選取在日期范圍內(nèi)的股票數(shù)據(jù)序列并按日期排序 stock_data = stock_data.ixstock_data'date'.isin(date), 'date', 'change', &#

4、39;adjust_price' stock_data = stock_data.ixstock_data'date'.isin(date), 'date', 'change', 'adjust_price' # 選取在日期范圍內(nèi)的指數(shù)數(shù)據(jù)序列并按日期排序 date_list = list(stock_data'date') benchmark = benchmark.ixbenchmark'date'.isin(date_list), 'date', 'change

5、', 'close' benchmark.sort_values(by='date', inplace=True) benchmark.set_index('date', inplace=True) # 將回測要用到的各個數(shù)據(jù)序列轉(zhuǎn)成list格式 date_line = list(benchmark.index.strftime('%Y-%m-%d') # 日期序列 capital_line = list(stock_data'adjust_price') # 賬戶價值序列 return_line = li

6、st(stock_data'change') # 收益率序列 indexreturn_line = list(benchmark'change') # 指數(shù)的變化率序列 index_line = list(benchmark'close') # 指數(shù)序列 return date_line, capital_line, return_line, index_line, indexreturn_line# 計算年化收益率函數(shù)def annual_return(date_line, capital_line): """ :

7、param date_line: 日期序列 :param capital_line: 賬戶價值序列 :return: 輸出在回測期間的年化收益率 """ # 將數(shù)據(jù)序列合并成dataframe并按日期排序 df = pd.DataFrame('date': date_line, 'capital': capital_line) df.sort_values(by='date', inplace=True) df.reset_index(drop=True, inplace=True) rng = pd.period_

8、range(df'date'.iloc0, df'date'.iloc-1, freq='D') # 計算年化收益率 annual = pow(df.ixlen(df.index) - 1, 'capital' / df.ix0, 'capital', 250 / len(rng) - 1 print '年化收益率為：%f' % annual# 計算最大回撤函數(shù)def max_drawdown(date_line, capital_line): """ :param d

9、ate_line: 日期序列 :param capital_line: 賬戶價值序列 :return: 輸出最大回撤及開始日期和結(jié)束日期 """ # 將數(shù)據(jù)序列合并為一個dataframe并按日期排序 df = pd.DataFrame('date': date_line, 'capital': capital_line) df.sort_values(by='date', inplace=True) df.reset_index(drop=True, inplace=True) df'max2here

10、9; = pd.expanding_max(df'capital') # 計算當(dāng)日之前的賬戶最大價值 df'dd2here' = df'capital' / df'max2here' - 1 # 計算當(dāng)日的回撤 # 計算最大回撤和結(jié)束時間 temp = df.sort_values(by='dd2here').iloc0'date', 'dd2here' max_dd = temp'dd2here' end_date = temp'date' # 計算

11、開始時間 df = dfdf'date' <= end_date start_date = df.sort_values(by='capital', ascending=False).iloc0'date' print '最大回撤為：%f, 開始日期：%s, 結(jié)束日期：%s' % (max_dd, start_date, end_date)# 計算平均漲幅def average_change(date_line, return_line): """ :param date_line: 日期序列

12、:param return_line: 賬戶日收益率序列 :return: 輸出平均漲幅 """ df = pd.DataFrame('date': date_line, 'rtn': return_line) ave = df'rtn'.mean() print '平均漲幅為：%f' % ave# 計算上漲概率def prob_up(date_line, return_line): """ :param date_line: 日期序列 :param return_lin

13、e: 賬戶日收益率序列 :return: 輸出上漲概率 """ df = pd.DataFrame('date': date_line, 'rtn': return_line) df.ixdf'rtn' > 0, 'rtn' = 1 # 收益率大于0的記為1 df.ixdf'rtn' <= 0, 'rtn' = 0 # 收益率小于等于0的記為0 # 統(tǒng)計1和0各出現(xiàn)的次數(shù) count = df'rtn'.value_counts() p_up

14、 = count.loc1 / len(df.index) print '上漲概率為：%f' % p_up# 計算最大連續(xù)上漲天數(shù)和最大連續(xù)下跌天數(shù)def max_successive_up(date_line, return_line): """ :param date_line: 日期序列 :param return_line: 賬戶日收益率序列 :return: 輸出最大連續(xù)上漲天數(shù)和最大連續(xù)下跌天數(shù) """ df = pd.DataFrame('date': date_line, 'rt

15、n': return_line) # 新建一個全為空值的series,并作為dataframe新的一列 s = pd.Series(np.nan, index=df.index) = 'up' df = pd.concat(df, s, axis=1) # 當(dāng)收益率大于0時，up取1，小于0時，up取0，等于0時采用前向差值 df.ixdf'rtn' > 0, 'up' = 1 df.ixdf'rtn' < 0, 'up' = 0 df'up'.fillna(met

16、hod='ffill', inplace=True) # 根據(jù)up這一列計算到某天為止連續(xù)上漲下跌的天數(shù) rtn_list = list(df'up') successive_up_list = num = 1 for i in range(len(rtn_list): if i = 0: successive_up_list.append(num) else: if (rtn_listi = rtn_listi - 1 = 1) or (rtn_listi = rtn_listi - 1 = 0): num += 1 else: num = 1 success

17、ive_up_list.append(num) # 將計算結(jié)果賦給新的一列'successive_up' df'successive_up' = successive_up_list # 分別在上漲和下跌的兩個dataframe里按照'successive_up'的值排序并取最大值 max_successive_up = dfdf'up' = 1.sort_values(by='successive_up', ascending=False)'successive_up'.iloc0 max_suc

18、cessive_down = dfdf'up' = 0.sort_values(by='successive_up', ascending=False)'successive_up'.iloc0 print '最大連續(xù)上漲天數(shù)為：%d 最大連續(xù)下跌天數(shù)為：%d' % (max_successive_up, max_successive_down)# 計算最大單周期漲幅和最大單周期跌幅def max_period_return(date_line, return_line): """ :param da

19、te_line: 日期序列 :param return_line: 賬戶日收益率序列 :return: 輸出最大單周期漲幅和最大單周期跌幅 """ df = pd.DataFrame('date': date_line, 'rtn': return_line) # 分別計算日收益率的最大值和最小值 max_return = df'rtn'.max() min_return = df'rtn'.min() print '最大單周期漲幅為：%f 最大單周期跌幅為：%f' % (max_re

20、turn, min_return)# 計算收益波動率的函數(shù)def volatility(date_line, return_line): """ :param date_line: 日期序列 :param return_line: 賬戶日收益率序列 :return: 輸出回測期間的收益波動率 """ from math import sqrt df = pd.DataFrame('date': date_line, 'rtn': return_line) # 計算波動率 vol = df'rtn

21、'.std() * sqrt(250) print '收益波動率為：%f' % vol# 計算貝塔的函數(shù)def beta(date_line, return_line, indexreturn_line): """ :param date_line: 日期序列 :param return_line: 賬戶日收益率序列 :param indexreturn_line: 指數(shù)的收益率序列 :return: 輸出beta值 """ df = pd.DataFrame('date': date_line

22、, 'rtn': return_line, 'benchmark_rtn': indexreturn_line) # 賬戶收益和基準(zhǔn)收益的協(xié)方差除以基準(zhǔn)收益的方差 b = df'rtn'.cov(df'benchmark_rtn') / df'benchmark_rtn'.var() print 'beta: %f' % b# 計算alpha的函數(shù)def alpha(date_line, capital_line, index_line, return_line, indexreturn_line)

23、: """ :param date_line: 日期序列 :param capital_line: 賬戶價值序列 :param index_line: 指數(shù)序列 :param return_line: 賬戶日收益率序列 :param indexreturn_line: 指數(shù)的收益率序列 :return: 輸出alpha值 """ # 將數(shù)據(jù)序列合并成dataframe并按日期排序 df = pd.DataFrame('date': date_line, 'capital': capital_line,

24、'benchmark': index_line, 'rtn': return_line, 'benchmark_rtn': indexreturn_line) df.sort_values(by='date', inplace=True) df.reset_index(drop=True, inplace=True) rng = pd.period_range(df'date'.iloc0, df'date'.iloc-1, freq='D') rf = 0.0284 # 無風(fēng)險利率

25、取10年期國債的到期年化收益率 annual_stock = pow(df.ixlen(df.index) - 1, 'capital' / df.ix0, 'capital', 250 / len(rng) - 1 # 賬戶年化收益 annual_index = pow(df.ixlen(df.index) - 1, 'benchmark' / df.ix0, 'benchmark', 250 / len(rng) - 1 # 基準(zhǔn)年化收益 beta = df'rtn'.cov(df'benchmark_

26、rtn') / df'benchmark_rtn'.var() # 計算貝塔值 a = (annual_stock - rf) - beta * (annual_index - rf) # 計算alpha值 print 'alpha：%f' % a# 計算夏普比函數(shù)def sharpe_ratio(date_line, capital_line, return_line): """ :param date_line: 日期序列 :param capital_line: 賬戶價值序列 :param return_line: 賬

27、戶日收益率序列 :return: 輸出夏普比率 """ from math import sqrt # 將數(shù)據(jù)序列合并為一個dataframe并按日期排序 df = pd.DataFrame('date': date_line, 'capital': capital_line, 'rtn': return_line) df.sort_values(by='date', inplace=True) df.reset_index(drop=True, inplace=True) rng = pd.peri

28、od_range(df'date'.iloc0, df'date'.iloc-1, freq='D') rf = 0.0284 # 無風(fēng)險利率取10年期國債的到期年化收益率 # 賬戶年化收益 annual_stock = pow(df.ixlen(df.index) - 1, 'close' / df.ix0, 'close', 250 / len(rng) - 1 # 計算收益波動率 volatility = df'rtn'.std() * sqrt(250) # 計算夏普比 sharpe = (

29、annual_stock - rf) / volatility print 'sharpe_ratio: %f' % sharpe# 計算信息比率函數(shù)def info_ratio(date_line, return_line, indexreturn_line): """ :param date_line: 日期序列 :param return_line: 賬戶日收益率序列 :param indexreturn_line: 指數(shù)的收益率序列 :return: 輸出信息比率 """ from math import sq

30、rt df = pd.DataFrame('date': date_line, 'rtn': return_line, 'benchmark_rtn': indexreturn_line) df'diff' = df'rtn' - df'benchmark_rtn' annual_mean = df'diff'.mean() * 250 annual_std = df'diff'.std() * sqrt(250) info = annual_mean / annua

31、l_std print 'info_ratio: %f' % info# 計算股票和基準(zhǔn)在回測期間的累計收益率并畫圖def cumulative_return(date_line, return_line, indexreturn_line): """ :param date_line: 日期序列 :param return_line: 賬戶日收益率序列 :param indexreturn_line: 指數(shù)日收益率序列 :return: 畫出股票和基準(zhǔn)在回測期間的累計收益率的折線圖 """ df = pd.DataFrame('date': date_line, 'rtn': return_line, 'benchmark_rtn': indexreturn_line) df'stock_cumret' = (df'rtn' + 1).cumprod() df'benchmark_cumret'