研報復現：技術分析算法、框架與實戰
《破解“看圖”之謎：技術分析算法、框架與實戰》是中泰證券于2021年8月發布的一份金融工程報告，旨在以科學方法系統化、自動化地識別技術圖形，并驗證其在實際市場中的有效性。
研報簡讀
一、研究背景與意義
技術分析長期被視為主觀性強、缺乏統一標準的分析方法。報告指出，其不被廣泛接受的主要原因有二：一是缺乏固定算法，導致不同分析者對同一圖表解讀不一；二是多數文獻未系統論證技術分析適用的市場環境與實踐效果。為此，報告借鑒Andrew W Lo等學者在2000年提出的方法，構建了一套科學、可重復的技術分析框架。
二、算法原理：核回歸與極值點識別
報告采用核回歸（Nadaraya-Watson估計） 對價格序列進行平滑處理，以消除噪聲并準確提取極值點。通過高斯核函數和交叉驗證法選擇最優帶寬，確保平滑效果既不過度也不不足。極值點通過導數符號變化識別，進而根據其排列與高低關系定義各類技術形態（如頭肩底、雙頂、三角形等）。
三、形態定義與識別
報告系統定義了幾類常見技術形態（頭肩頂/底、發散形態、三角形、矩形、雙頂/底），并通過五個連續極值點的相對位置和幅度關系進行數學刻畫。識別過程包括：平滑價格序列 → 求導找極值 → 按定義匹配形態。
四、實證分析：頭肩底形態的有效性
報告重點驗證了“頭肩底（IHS）”形態在行業指數和寬基指數上的表現（2010–2021年數據），發現：
- 在指數層面，多數行業在形態出現后10日內獲得正收益，勝率普遍超過50%，部分行業（如化工、汽車、電氣設備）勝率可達70%以上。
- 結合均線條件（如MA5>MA20） 可進一步提升勝率。
- 在個股層面，頭肩底形態表現不穩定，勝率分布接近隨機（以50%為中位數），提示技術分析在個股擇時中需謹慎使用。
五、結論與風險提示
技術分析在指數擇時中具有顯著價值，尤其在中期（10日）展望中表現穩健；但在個股層面效果有限，需結合其他分析方法。報告強調，所有結論基于歷史數據統計，存在滯后性、模型局限性和市場環境變化風險，投資者應謹慎參考。
總結
該報告通過算法化、系統化的方法，為技術分析提供了科學的實證基礎，打破了其“主觀藝術”的傳統印象，尤其在指數投資中展現出實用價值，為量化投資與技術分析的結合提供了重要參考。
代碼復現
論文介紹：《Foundations of Technical Analysis》
Andrew W Lo、Harry Mamaysky和王江提出了一種系統化和自動化的方法，使用非參數核回歸方法來進行模式識別，并將該方法應用于1962年至 1996 年的美股數據，以評估技術分析的有效性。通過將股票收益的無條件經驗分布與條件分布（給定特定的技術指標，例如：頭肩底形態、雙底形態）作比較，發現在 31 年的樣本期 間內，部分技術指標確實提供了有用的信息，具有實用價值。 與基本面分析不同，技術分析一直以來飽受爭議。
然而一些學術研究表明，技術分析能從市場價格中提取 有用的信息。例如，Lo and MacKinlay (1988, 1999)證實了每周的美股指數并非隨機游走，過去的價格可以在某種 程度上預測未來收益。技術分析和傳統金融工程的一個重要區別在于，技術分析主要通過觀察圖表進行，而量化金融則依賴于相對完善的數值法。因此，技術分析利用幾何工具和形態識別，而量化金融運用數學分析和概率統計。
隨著近年來金融工程、計算機技術和數值算法等領域的突破，金融工程可以逐步取代不那么嚴謹的技術分析。技術分析雖飽受質疑卻仍能占據一席之地，歸功于其視覺分析模式更貼近直觀認知，而且在過去，要進行技術分析，首先要認識到價格過程是非線性的，且包含一定的規律和模式。為了定量地捕捉這種規 律，我們首先假定價格過程尚未有嚴格的算法取代傳統的技術分析，如今，成熟的統計算法能夠取代傳統的幾何畫圖，讓技術分析繼續以 更新、更嚴謹的方式服務投資者，同時金融工程領域在分析范式上也得到了豐富。
形態識別算法
核估計方法
假定價格過程Pt有如下表達形式:
其中Xt是狀態變量，m(Xt)是任意固定但位置的非線性函數，
為白噪聲。
為了識別模式,我們令狀態變量等于時間Xt = t，此時,需要用一個光滑函數
。近似價格過程為了與核回歸估計文獻中的符號保持一致，我們仍將狀態變量記為Xt。
還需要一個形態識別的算法,以自動識別指數指標。一旦有了算法,就可以應用于不同時段的資產價格,從而評估不同技術指標的有效性。
平滑估計量
其中離x較近的Xt對應的Pt擁有較大的權重
。對于距離的選擇，太寬會導致估計量過于平滑而無法顯示出
真正的特性，太窄又會導致估計量的波動較大，無法排除噪聲的影響。因此需要通過選擇合適的權重
來平衡以上兩點。
核回歸
形態識別算法
1、頭肩形態（頭肩頂和頭肩底）
2、發散形態（頂部發散和底部發散）
3、三角形
4、矩形
形態識別測試
中國寶安
data1 = get_price('000009.XSHE', start_date='2021-01-21', end_date='2021-12-31',
fields=['open', 'close', 'low', 'high'], panel=False)
patterns_record1 = rolling_patterns2pool(data1['close'],n=35)
plot_patterns_chart(data1,patterns_record1,True,False)
plt.title('中國寶安')
plot_patterns_chart(data1,patterns_record1,True,True);
江淮汽車
data2 = get_price('600418.XSHG', start_date='2021-01-21', end_date='2021-12-31',
fields=['open', 'close', 'low', 'high'], panel=False)
patterns_record2 = rolling_patterns2pool(data1['close'],n=35)
plot_patterns_chart(data1,patterns_record2,True,False)
plt.title('江淮汽車')
plot_patterns_chart(data1,patterns_record2,True,True);
滬深300
hs300 = get_price('000300.XSHG', start_date='2014-01-01', end_date='2021-12-31',
fields=['open', 'close', 'low', 'high'], panel=False)
# 這里周期周期較長 加入reset_window設置字典更新頻率
patterns_record3 = rolling_patterns2pool(hs300['close'],n=35,reset_window=120)
plot_patterns_chart(hs300,patterns_record3,True,False)
plt.title('滬深300')
plot_patterns_chart(hs300,patterns_record3,True,True);
申萬一級行業形態識別情況
def patterns_res2json(dic: Dict) -> str:
"""將結果轉為json
Args:
dic (Dict): 結果字典
Returns:
str
"""
def json_serial(obj):
"""JSON serializer for objects not serializable by default json code"""
if isinstance(obj, np.datetime64):
return pd.to_datetime(obj).strftime('%Y-%m-%d')
if isinstance(obj, np.ndarray):
return list(map(lambda x: pd.to_datetime(x).strftime('%Y-%m-%d'), obj.tolist()))
return json.dumps(dic, default=json_serial, ensure_ascii=False)
def pretreatment_events(factor: pd.DataFrame, returns: pd.DataFrame, before: int, after: int) -> pd.DataFrame:
"""預處理事件,將其拉到同一時間
Args:
factor (pd.DataFrame): MuliIndex level0-date level1-asset
returns (pd.DataFrame): index-datetime columns-asset
before (int): 事件前N日
after (int): 事件后N日
Returns:
pd.DataFrame: [description]
"""
all_returns = []
for timestamp, df in factor.groupby(level='date'):
equities = df.index.get_level_values('asset')
try:
day_zero_index = returns.index.get_loc(timestamp)
except KeyError:
continue
starting_index = max(day_zero_index - before, 0)
ending_index = min(day_zero_index + after + 1,
len(returns.index))
equities_slice = set(equities)
series = returns.loc[returns.index[starting_index:ending_index],
equities_slice]
series.index = range(starting_index - day_zero_index,
ending_index - day_zero_index)
all_returns.append(series)
return pd.concat(all_returns, axis=1)
def get_event_cumreturns(pretreatment_events: pd.DataFrame) -> pd.DataFrame:
"""以事件當日為基準的累計收益計算
Args:
pretreatment_events (pd.DataFrame): index-事件前后日 columns-asset
Returns:
pd.DataFrame
"""
df = pd.DataFrame(index=pretreatment_events.index,
columns=pretreatment_events.columns)
df.loc[:0] = pretreatment_events.loc[:0] / pretreatment_events.loc[0] - 1
df.loc[1:] = pretreatment_events.loc[0:] / pretreatment_events.loc[0] - 1
return df
def get_industry_price(codes: Union[str, List], start: str, end: str) -> pd.DataFrame:
"""獲取行業指數日度數據. 限制獲取條數Limit=4000
Args:
codes (Union[str,List]): 行業指數代碼
start (str): 起始日
end (str): 結束日
Returns:
pd.DataFrame: 日度數據
"""
def query_func(code: str, start: str, end: str) -> pd.Series:
return finance.run_query(query(finance.SW1_DAILY_PRICE).filter(finance.SW1_DAILY_PRICE.code == code,
finance.SW1_DAILY_PRICE.date >= start,
finance.SW1_DAILY_PRICE.date <= end))
if isinstance(codes, str):
codes = [codes]
return pd.concat((query_func(code, start, end) for code in codes))
def calc_events_ret(ser: pd.Series, pricing: pd.DataFrame, before: int = 3, end: int = 10, group: bool = True) -> Union[pd.Series, pd.DataFrame]:
""" 計算形態識別前累計收益率情況
Args:
ser (pd.Series): _description_
pricing (pd.DataFrame): 價格數據 index-date columns-指數
before (int, optional): 識別前N日. Defaults to 3.
end (int, optional): 識別后N日. Defaults to 10.
group (bool, optional): 是否分組. Defaults to True.
Returns:
Union[pd.Series, pd.DataFrame]
"""
events = pretreatment_events(ser, pricing, before, end)
rets = get_event_cumreturns(events)
if group:
return rets.mean(axis=1)
else:
return rets
def get_win_rate(df: pd.DataFrame) -> pd.DataFrame:
"""計算勝率
Args:
df (pd.DataFrame): index-days columns
Returns:
pd.DataFrame
"""
return df.apply(lambda x: np.sum(np.where(x > 0, 1, 0)) / x.count(), axis=1)
def get_pl(df:pd.DataFrame)->pd.DataFrame:
"""計算盈虧比
Returns:
pd.DataFrame
"""
return df.apply(lambda x:x[x>0].mean() / x[x<0].mean(),axis=1)
def plot_events_ret(ser: pd.Series, title: str = '', ax=None):
"""繪制事件收益率圖
Args:
ser (pd.Series): 收益率序列
ax (_type_, optional):Defaults to None.
Returns:
ax
"""
if ax is None:
fig, ax = plt.figure(figsize=(18, 4))
line_ax = ser.plot(ax=ax, marker='o', title=title)
line_ax.yaxis.set_major_formatter(
mpl.ticker.FuncFormatter(lambda x, pos: '%.2f%%' % (x * 100)))
line_ax.set_xlabel('天')
line_ax.set_ylabel('平均收益率')
ax.axvline(0, ls='--', color='black')
return ax
# 獲取申萬一級行業列表
indstries_frame = get_industries(name='sw_l1', date=None)
industry_price = get_industry_price(indstries_frame.index.tolist(),'2014-01-01','2022-02-18')
# 數據儲存
industry_price.to_csv('sw_lv1.csv')
# 讀取申萬一級行業數據
industry_price = pd.read_csv('sw_lv1.csv', index_col=[
'name', 'date'], parse_dates=True).drop(columns='Unnamed: 0')
industry_price.head()
# time 2:14:46
## 形態識別數量受窗口期 及 更新字典的窗口 大小影響
dic = {} # 儲存形態識別結果
for name,df in tqdm(industry_price.groupby(level='name')):
if len(df) > 120:
dic[name] = rolling_patterns2pool(df.loc[name,'close'],35,reset_window = 120)._asdict()
# 將結果儲存為json
res_json =patterns_res2json(dic)
# 數據儲存
with open('res_json.json','w',encoding='utf-8') as file:
json.dump(res_json,file)
# 讀取 形態識別后的文件
with open('res_json.json','r',encoding='utf-8') as file:
res_json = json.load(file)
# 獲取交易日歷
trade_calendar = get_trade_days('2013-06-01','2022-02-21')
idx = pd.to_datetime(trade_calendar)
row_data = [] # 獲取形態識別時的時點
res_dic = json.loads(res_json) # json轉為字典
for code,res1 in res_dic.items():
for pattern_name,point_tuple in res1['patterns'].items():
for p1,p2 in point_tuple:
watch_date = idx.get_loc(p2) + 3 # 模擬三天后識別形態
row_data.append([code,pattern_name,idx[watch_date]])
# 轉為frame格式
stats_df = pd.DataFrame(row_data,columns=['指數','形態','時間'])
stats_df['value'] = 1
factor_df = pd.pivot_table(stats_df,index=['時間','指數'],columns=['形態'],values='value')
factor_df = factor_df.sort_index()
factor_df.index.names = ['date','asset']
pricing = pd.pivot_table(industry_price.reset_index(),
index='date', columns='name', values='close')形態識別后，前后平均收益情況
# TODO:收益減去指數自身 評價其超額情況 否則無法真實評價
group_ret = factor_df.groupby(level=0, axis=1).apply(
lambda x: calc_events_ret(x.dropna(), pricing))
size = group_ret.shape[1]
fig, axes = plt.subplots(size, figsize=(18, 4 * size))
axes = axes.flatten()
for ax, (name, ser) in zip(axes, group_ret.items()):
plot_events_ret(ser, name, ax)
plt.subplots_adjust(hspace=0.4)
# 計算勝率
evet_ret = factor_df.groupby(level=0, axis=1).apply(
lambda x: calc_events_ret(x.dropna(), pricing, group=False))
grouped = evet_ret.groupby(level=[0, 1], axis=1)
# 計算勝率
win_ratio = grouped.apply(get_win_rate).loc[[3, 5, 10]].T.swaplevel().sort_index()
# 計算盈虧比
pl_df = grouped.apply(get_pl).loc[[3, 5, 10]].T.swaplevel().sort_index()
win_ratio.columns = pd.MultiIndex.from_tuples([('勝率',3),('勝率',5),('勝率',10)])
pl_df.columns = pd.MultiIndex.from_tuples([('盈虧比',3),('盈虧比',5),('盈虧比',10)])
pattern_count = factor_df.groupby(level=1).sum().stack()
stats = pd.concat((win_ratio,pl_df),axis=1)
stats[('識別次數','All')] = pattern_count
圖片
本文轉載自??靈度智能??，作者：靈度智能
