📢 レイヤードアーマー
この記事は実用的取引のためのウェーブレット変換に関するポピュラーサイエンスこのコードは、（マルチレベル分解、しきい値ノイズ除去、標準ウェーブレットの逆変換再構成などの複雑な手順を省略した）簡略化された教育バージョンであり、コアとなるアイデアのみを保持しています。**ウェーブレット係数を使用して価格のマルチスケール平滑化を実行し、トレンド情報を抽出します。**戦略開発と迅速な検証には適していますが、学術研究や論文の出版には適していません。
I. はじめに：「知虎達人」（専門家）の虚偽を暴く
Zhihu で金融や定量的なトピックを頻繁に閲覧している人は、きっとこのシナリオを見たことがあるでしょう。
一部の「専門家」は次のように言い続けています。
- ウェーブレット変換ノイズ低減
- フーリエ変換抽出期間
- ラプラス平滑化により外れ値が除去されます。
彼はまるで定量取引の核兵器を習得したかのように、皆を驚かせた。
でも、彼にコードを見せてもらいたいんですか?
「これは…企業秘密なので、公開できません。」
彼にその原理を説明するように頼んでください。
「これは…高度な数学が絡んでくるので、説明しても理解できないと思いますよ。」
今日は、これらの「Zhihu の専門家」が頻繁に言及するトピックを探り、金融市場におけるウェーブレット変換の実際のアプリケーションを紹介し、誰もがこの技術を正しく理解できるようにお手伝いします。
II. ウェーブレット変換とは何でしょうか?
わかりやすい説明
曲を聴いているときに、録音にバックグラウンド ノイズがあると想像してください。
オリジナル録音 = 人間の声 + 背景ノイズ + 雑音
ウェーブレット変換はスマートフィルター：
- 声を守る
- ノイズを除去する
- どの部分がコーラスで、どの部分がヴァースなのかもわかります。
金融市場に切り替える:
元の価格 = 真のトレンド + 短期的な変動 + ランダムノイズ
ウェーブレット変換は次のことに役立ちます:
- 真のトレンドの抽出（長期的な方向性）
- 短期的な変動をフィルタリングする（日中変動）
- 重要な変曲点を特定する（トレンド反転）
コアコンセプト：基底関数分解
ウェーブレット変換の本質は特定の「基底関数」（ウェーブレット）のセットを使用して元の信号を分解します。。
ある人物の外見を説明したいとします。
- 従来の方法: ピクセルごとに記述しますが、非常に面倒です。
- ウェーブレット法：「目の大きさ、鼻の高さ、顔の輪郭」などの要素を使用する特徴説明を組み合わせる
金融価格において：
元の価格系列 = 基底関数 1 × 重み 1 + 基底関数 2 × 重み 2 + ... + ノイズ
**基底関数は、ウェーブレット係数に対応する「テンプレート」です。**異なるウェーブレット タイプ (Haar、Daubechies、Mexican Hat など) では、価格を分解するために異なる「特徴抽出器」を使用するのと同じように、異なるテンプレートが使用されます。
フィルタ: 周波数領域におけるふるい
ウェーブレット変換は本質的には...マルチスケールフィルタバンク:
高周波フィルタ → 急激な変動（日々のノイズ、ティックレベルの変動）を捕捉
中周波フィルター → 中期的なトレンド（数時間から数日の範囲）を捉えます。
低頻度フィルター → 長期的な傾向（週次および月次傾向）を捉えます。
なぜ「ウェーブレット」と呼ばれるのでしょうか?
- 従来のフーリエ変換では無限に長い正弦波無限に長い定規のように
- ウェーブレット変換が使用される有限長の「小さな」波異なる長さの定規のセットのように
正弦波を使って金融価格を分析する際の問題点：正弦波は信号が周期的に繰り返されると想定しますが、金融市場はそうではありません！BTCは今日10%上昇し、明日は8%下落する可能性があり、周期性は全くありません。
ウェーブレットの利点:ローカリゼーション分析「相場全体が変動していた」といった一般的な結論ではなく、「2025年12月20日午後3時から午後5時の間は、価格の傾向は主に上昇していた」ということが分かります。
再構築：分解から再構築へ
ウェーブレット変換は可逆これはとても重要です！
元の価格 ---> ウェーブレット分解 ---> トレンド成分 + ボラティリティ成分 + ノイズ成分
トレンド成分 + ボラティリティ成分 + ノイズ成分 ---> ウェーブレット再構成 ---> 元の価格
リファクタリングプロセス個々のコンポーネントを分解することです。選択的に結合：
- トレンド成分 =[99800, 99850, 99900, 99950, ...] # 欲しいもの
- 変動成分 =[+200、-150、+180、-120、...] # 役に立つかもしれない
- ノイズ成分 =[±10、±15、±8、±12、...] # 破棄!
再構築中はトレンドコンポーネントのみを使用する
分解後、以下のものが得られました
- 再構築価格 = トレンド成分 + 部分的ボラティリティ成分
実際の取引では、通常低周波部分のみを再構築する（トレンド）高周波成分（ノイズ）はそのまま除去されます。これがウェーブレット「ノイズリダクション」の原理です。
数学の原理（簡易版）
複雑な積分式を省略して、一般の人にもわかるように説明しましょう。
ウェーブレット変換 = 「ウェーブレット係数」のセットを使用した価格系列の加重平均
基本式:
価格の平滑化[i] = Σ(元の価格)[i-j] × ウェーブレット係数[j]) / Σ(ウェーブレット係数)[j])
フィルターの視点：
元の価格はウェーブレット フィルターを介してフィルタリングされ、異なる周波数のコンポーネントが「選択」されます。
鍵はウェーブレット係数の選択：
- 異なるウェーブレット = 異なるフィルタ特性 = 異なる周波数応答
- 異なるレベル = 異なる時間スケール = 異なるトレンドサイクル
例えば：
Daubechies 4ウェーブレットを使用すると仮定すると、係数は[0.483, 0.837, 0.224, -0.129]:
この係数のセットはフィルタを定義します。
- 正の係数（0.483、0.837、0.224）→これらのポジションの価格を保持します。
- 負の係数（-0.129）→以前の価格変動を抑制する
- 係数加重 → 各価格の寄与を決定します。
このフィルターを価格系列全体に「スライド」させることで、ウェーブレット変換が完了します。スライドごとに計算が行われます。現在のウィンドウ内の価格の加重平均重みはウェーブレット係数です。
なぜ信号を「分解」できるのでしょうか?
なぜなら、次のことが数学的に証明できるからです。**あらゆる信号は、ウェーブレット基底関数の線形結合として表すことができます。**RGBの三原色を混ぜ合わせることであらゆる色を作り出すことができるのと同様に、ウェーブレット基底関数を組み合わせることであらゆる価格系列を導き出すことができます。ウェーブレットの種類によって、異なる信号分析に適した「基底関数ライブラリ」が提供されます。
III. この試み：7つのウェーブレット変換の実用化
応用アプローチ：理論から実践への簡素化
信号処理の教科書では、ウェーブレット変換には通常、複雑な処理が含まれます...多段階分解、再構成、閾値ノイズ除去手順:
完全なウェーブレット解析ワークフロー:
- マルチスケール分解 → 近似係数と詳細係数を生成します。
- しきい値設定 → ノイズを低減するための詳細係数のソフト/ハードしきい値設定
- 逆変換再構成 → 処理された係数を信号に復元する
- 境界拡張 → 信号境界効果の処理
- エネルギー正規化 → 変換前後のエネルギー保存を保証する
しかし金融取引の実用化中国では、そこまで複雑になる必要はありません。なぜなら：
1. 取引にはトレンドの方向だけが必要であり、完璧な再構築は必要ありません。
学術研究では0.01%未満の再構築誤差が求められる場合がありますが、取引においては価格が上昇するか下落するかを判断するには十分です。再構築に5%の誤差があっても、トレンドの方向が正しければ、この戦略は依然として利益を上げることができます。
2. リアルタイム要件により計算が簡素化されます。
完全なウェーブレット分解には、複数の係数層の再帰計算が必要であり、高頻度取引では遅延が発生する可能性があります。一方、直接畳み込みは数ミリ秒で完了するため、ライブ取引のニーズを満たします。
3. 金融シグナルの特殊性
金融価格は安定したシグナルではなく、厳密な周期性も示しません。複雑な頻度分解はここではあまり意味がなく、単純なトレンド抽出の方が実用的です。
この簡素化戦略
したがって、この記事ウェーブレット変換の本質を抽出する金融市場の最も実践的な側面に焦点を当てます。
コア簡素化1: 近似係数のみを使用する（低頻度の傾向）
従来のウェーブレット分解 → 近似係数 + 詳細係数（多層）
このアプリケーション：近似係数のみを保持 → 平滑化されたトレンドを直接取得します。
コアの簡素化2: 閾値処理とノイズ除去を行わない直接畳み込み
従来のウェーブレット分解 → 詳細係数の閾値設定 → 再構成
このアプリケーション: 直接畳み込み → 平滑化された価格を取得する
コアの簡素化3: 境界処理を無視する
従来のウェーブレットでは、信号境界の対称拡張や周期拡張などの処理が必要です。
このアプリケーションは中間セクションのみに焦点を当てており、境界エラーは許容されます。
実装方法: フィルタ畳み込み
python
def convolve(src, coeffs, step):
"""
核心算法:用小波系数对价格序列做加权平均
src: 价格序列 [100000, 101000, 99000, ...]
coeffs: 小波系数 [0.483, 0.837, 0.224, -0.129]
step: 采样步长(用于多层级)
"""
sum_val = 0.0 # 加权和
sum_w = 0.0 # 权重和
for i, weight in enumerate(coeffs):
idx = i * step
if idx < len(src):
sum_val += src[idx] * weight
sum_w += weight
return sum_val / sum_w # 归一化
この機能は...ウェーブレットフィルタの核：
- 各ローソク足について、N 本のローソク足を調べます (N = ウェーブレット係数の数)。
- ウェーブレット係数を重みとして使用して加重平均を計算します。
- 調整することで
stepパラメータにより、複数レベルのスムージング（レベル 1/2/3...）が可能になります。
なぜこの単純化が合理的なのでしょうか?
取引の基本的な要件は次のとおりです。ノイズの傾向を見つけるウェーブレット変換の近似係数は、それ自体が信号の「ローパス フィルター」であり、低周波のトレンド成分を保持します。これはまさに必要なことです。
完全なウェーブレット解析の方がより正確ですが、金融取引においては、
- 利益はトレンドの方向から生まれるそれは復元の正確さから来るものではありません。
- よりシンプルな方法はより堅牢である複雑なモデルは過剰適合になりやすい
- 計算速度の方が重要実際の取引では、1 ミリ秒ごとにお金が動きます。
データ収集：FMZプラットフォームの利便性
使用Inventor Quantization（FMZ）プラットフォームのローカルバックテストエンジンデータを取得するのに非常に便利です！
python
'''backtest
start: 2025-12-17 00:00:00
end: 2025-12-23 08:00:00
period: 1h
basePeriod: 1h
exchanges: [{"eid":"Futures_Binance","currency":"BTC_USDT","fee":[0,0]}]
'''
from fmz import *
task = VCtx(__doc__)
def main():
exchange.SetCurrency("BTC_USDT")
exchange.SetContractType("swap")
records = exchange.GetRecords(PERIOD_H1, 500)
return records
records = main()
複雑なAPI統合やデータクレンジングは不要で、標準化されたローソク足データを直接取得できます。これにより、データ処理に煩わされることなく、7種類のウェーブレットの実際の効果を迅速に検証できます。
テスト対象
暗号通貨の価格における 7 つの一般的なウェーブレット タイプ (Haar、Daubechies 4、Symlet 4、Biorthogonal 3.3、Mexican Hat、Morlet、および Discrete Meyer) のパフォーマンスを比較すると、次の視覚的なデモンストレーションが提供されます。
- 異なるウェーブレット間の平滑化強度の違い
- 同じウェーブレットの異なるレベルでの効果の変化
- どのウェーブレットが短期取引に適しており、どのウェーブレットがトレンドフォローに適していますか?
**焦点は数学的導出の厳密さではなく、実際の結果の視覚的な比較にあります。**これにより、トレーダーは直感的な理解を深め、自分の戦略に適したウェーブレット タイプを選択できるようになります。
IV. 7種類のウェーブレットの詳細な説明
1. Haar ウェーブレット - 最も単純な平均
Haar ウェーブレットは、係数が 2 つだけの最も基本的なタイプのウェーブレットです。[0.5, 0.5]本質的には、隣接する 2 つの価格の単純平均です。
コアコード:
python
coeffs = [0.5, 0.5]
# 对价格序列 [100000, 101000, 99000, 102000, 98000] 处理
def smooth(prices, i):
return (prices[i] * 0.5 + prices[i-1] * 0.5) / 1.0
# 结果：[100000, 100500, 100000, 100500, 100000]
ご覧のとおり、当初は99,000から102,000まで大きく変動していた価格が、Haar処理によって比較的安定しています。これはウェーブレット「ノイズ除去」の効果です。つまり、短期的な激しい変動を平滑化し、より滑らかな価格トレンドを観察できるようになります。
2. Daubechies 4 - エンジニアリングプロジェクトでよく使用される
Daubechies 4（略してdb4）は、工学分野で最もよく使われるウェーブレットの一つです。その係数は以下のとおりです。[0.483, 0.837, 0.224, -0.129]最後の係数は...であることに注意してください。負の数それがユニークな点です。
コアコード:
python
coeffs = [0.483, 0.837, 0.224, -0.129]
# 处理第i个价格点
def smooth(prices, i):
weighted_sum = (prices[i] * 0.483 + # 当前价格
prices[i-1] * 0.837 + # 前1根，权重最大！
prices[i-2] * 0.224 + # 前2根
prices[i-3] * (-0.129)) # 前3根，负权重
weight_sum = 0.483 + 0.837 + 0.224 + (-0.129) # = 1.415
return weighted_sum / weight_sum
# 示例：smooth([100000, 101000, 99000, 102000], 3) ≈ 100251
**主な機能:**前のローソク足の重み（0.837）が現在の価格の重み（0.483）よりも大きいです。これは、db4が「直前に発生した価格」をより重視し、負の重み係数が以前の価格を「相殺」する効果をもたらし、滑らかさをさらに高めることを意味します。
3. Symlet 4 - 対称的な改良版
Symlet 4はDaubechiesの改良版であり、より対称性を高めることを目指しています。係数：[-0.076, -0.030, 0.498, 0.804, 0.298, -0.099, -0.013, 0.032]。
コアコード:
python
coeffs = [-0.076, -0.030, 0.498, 0.804, 0.298, -0.099, -0.013, 0.032]
# 向前看8根K线
def smooth(prices, i):
weighted_sum = sum(prices[i-j] * coeffs[j] for j in range(8))
weight_sum = sum(coeffs)
return weighted_sum / weight_sum
# 平滑效果比Haar和db4都强，但反应速度更慢
**主な機能:**8本のローソク足のウィンドウ長は、価格の「記憶」をより長く保つことを可能にします。滑らかな曲線では、真のトレンド反転は8本のローソク足が経過するまで確認できない場合があります。
4. 双直交3.3 - 完全な対称性
双直交 3.3 (略称 bior3.3) は、次の係数を持つ完全に対称なウェーブレットです。[-0.066, 0.283, 0.637, 0.283, -0.066]。
コアコード:
python
coeffs = [-0.066, 0.283, 0.637, 0.283, -0.066]
# ↑ 中心↑ ↑
# 完全对称的两端
# 处理中间价格点
def smooth(prices, i):
# 实际应用：只向前看，不使用未来数据
weighted_sum = (prices[i-4] * (-0.066) + # 前4根
prices[i-3] * 0.283 + # 前3根
prices[i-2] * 0.637 + # 前2根，权重最大
prices[i-1] * 0.283 + # 前1根
prices[i] * (-0.066)) # 当前
weight_sum = sum(coeffs) # = 1.071
return weighted_sum / weight_sum
**主な機能:**対称性により、「位相歪み」がなくなるため、滑らかな曲線が不自然に左または右にシフトすることはありません。
5. メキシカンハット - ターニングポイントハンター
メキシカンハット（リッカーウェーブレットとも呼ばれる）係数:[-0.1, 0.0, 0.4, 0.8, 0.4, 0.0, -0.1]メキシコのソンブレロのような形をしています。
コアコード:
python
coeffs = [-0.1, 0.0, 0.4, 0.8, 0.4, 0.0, -0.1]
# 负值 零 正值 最大 正值 零 负值
# ↓ ↓
# "惩罚"两端，增强拐点检测能力
def smooth(prices, i):
weighted_sum = (prices[i-6] * (-0.1) + # 左3，负权重
prices[i-5] * 0.0 + # 左2
prices[i-4] * 0.4 + # 左1
prices[i-3] * 0.8 + # 中心，权重最大
prices[i-2] * 0.4 + # 右1
prices[i-1] * 0.0 + # 右2
prices[i] * (-0.1)) # 右3，负权重
weight_sum = sum(coeffs)
return weighted_sum / weight_sum
主な機能:「中央が大きく、両端がマイナス」の構造のため、検出に特に優れています。変曲点- 価格が上昇トレンドから下降トレンドへ（あるいはその逆）反転する重要な瞬間。負の加重係数は、遠い価格に「ペナルティ」を課し、トレンドの変化を素早く捉えます。
6. モルレ - ガウス平滑化の王
モルレウェーブレットはガウス（正規）分布に基づいており、係数は次のようになります。[0.0625, 0.25, 0.375, 0.25, 0.0625]。
コアコード:
python
coeffs = [0.0625, 0.25, 0.375, 0.25, 0.0625]
# ↓ ↓ ↓中心 ↓ ↓
# 远端 近端 最高 近端 远端
# 完美的高斯钟形曲线
def smooth(prices, i):
weighted_sum = (prices[i-4] * 0.0625 + # 左2，6.25%
prices[i-3] * 0.25 + # 左1，25%
prices[i-2] * 0.375 + # 中心，37.5%
prices[i-1] * 0.25 + # 右1，25%
prices[i] * 0.0625) # 右2，6.25%
# 权重和正好 = 1.0，无需除法
return weighted_sum
**主な機能:**すべてのウェーブレットの中で最も「穏やか」なウェーブレットで、負の重み付けがなく、すべての価格が計算に穏やかに組み込まれます。結果として得られる曲線は非常に滑らかですが、応答が遅いという欠点があります。急激な価格変動は、数本のローソク足が経過するまで反映されない可能性があります。
7. ディスクリートマイヤー - 究極のスムージング
離散マイヤーウェーブレットは、次のような係数を持つ最も複雑なウェーブレットです。[-0.015, -0.025, 0.0, 0.28, 0.52, 0.28, 0.0, -0.025, -0.015]。
コアコード:
python
coeffs = [-0.015, -0.025, 0.0, 0.28, 0.52, 0.28, 0.0, -0.025, -0.015]
# ↑ ↑ ↑ ↑中心↑ ↑ ↑ ↑
# 完全对称，中心权重超过50%
def smooth(prices, i):
# 向前看9根K线
weighted_sum = sum(prices[i-j] * coeffs[j] for j in range(9))
weight_sum = sum(coeffs) # = 1.0
return weighted_sum
# 注意：第4根之前的K线权重是0.52，超过50%！
# 实际上在告诉你"4根K线之前的中期趋势"
**主な機能:**係数が最も多く（9）、ヒストリーデータが最も長く、平滑化効果が最も強い。「週次トレンド」の抽出には適しているが、遅延が大きく、価格が10%下落したとしても、曲線は依然として「上昇継続」を示す可能性がある。
5. スムージング効果に違いがあるのはなぜですか?
7 種類のウェーブレットを確認した後、次のパターンに気付いたはずです。
係数が多い = より遠くまで見える = より強い平滑化 = より大きな遅延
Haar (2係数) → 1つのバーだけを見る → ほとんど滑らかではない
ドーブシー 4 (4個) → 3つ前を参照 → やや滑らか
メキシカンハット（7）→6を参照→中程度のスムージング
離散メイヤー（9）→8本のバーを見る前→強めの平滑化
負の重みの効果は、感度を高め、変化を検出しやすくなることです。
Haar/Morlet（負の重みなし）→穏やかで滑らか、鈍感
メキシカンハット（両端が負）→変曲点に敏感
ドーブシーズ 4（ネガティブ）→トレンドの変化に敏感
対称性の役割 = 歪みがない = 元の形状を維持する
非対称性（ドーブシー）→左/右にずれる場合があります
対称性（双直交/マイヤー）→中心位置の維持
VI. 平滑化レベルの影響
ウェーブレット変換は、入れ子人形のように再帰的に適用できます。最初の適用はレベル1と呼ばれ、レベル1の結果に再度適用したものはレベル2と呼ばれ、以下同様に続きます。
さまざまなレベルで見られる時間スケール:
BTC取引に1時間足ローソク足チャートを使用すると仮定します。
レベル1 → 2～4時間にわたる短期的な変動を観察する
レベル2 → 4～8時間にわたって傾向を観察する
レベル3 → 1～2日間の中期トレンドを観察する（よく使われる戦略）
レベル4 → 2～4日間の価格帯を観察する
レベル5 → 4～8日間の主要な傾向を観察する
実際の結果の比較:
元のBTC価格（1時間チャート）:99500, 99800, 99200, 100200, 99800, 100500, 100100, ...
レベル 1 処理: 99600、99650、99500、99900、99950、100200、100250、...
（少し平滑化されていますが、変動はまだ見られます）
レベル 3 処理: 99620、99650、99700、99800、99950、100100、100200、...
（平滑化されており、中期的な傾向を示している）
レベル 5 処理: 99630、99640、99660、99700、99760、99840、99930、...
（非常にスムーズで、大まかな方向のみを示しています）
選択の原則はシンプルです。保有期間に基づいて対応するレベルを使用します。
15分スキャルピング → レベル1-2
日中取引 → レベル2-3
数日間のスイング→レベル3～4
長期トレンド分析 → レベル4-5
VII. 実践的な応用アイデア
ウェーブレット変換を取引に応用するのは非常に簡単です。ウェーブレット変換によって生成される平滑化された価格曲線を用いてトレンドの方向を判断し、トレンドが変化した時に取引を行うのです。具体的には、平滑化された終値が前回よりも高い場合は上昇トレンドを示しているため、ロングポジションを取り、平滑化された終値が前回よりも低い場合は下降トレンドを示しているため、ポジションを決済するかショートポジションを取ります。このロジックが効果的なのは、ウェーブレット変換によって短期的なランダムな変動が除去され、「上昇」または「下降」がノイズによる誤ったシグナルではなく、真のトレンド変化の可能性が高いと判断されるためです。
python
# 执行小波变换
transformed = transformer.transform_ohlc(df)
# 获取最近两根K线的平滑收盘价
w_close_current = transformed['w_close'].values[-1] # 当前平滑收盘价
w_close_prev = transformed['w_close'].values[-2] # 前一根平滑收盘价
# 判断趋势方向
signal = 0
if w_close_current > w_close_prev:
signal = 1 # 平滑价格向上 → 做多
elif w_close_current < w_close_prev:
signal = -1 # 平滑价格向下 → 做空
# 获取账户信息
account = exchange.GetAccount()
ticker = exchange.GetTicker()
if not account or not ticker:
Log("[Warning] Failed to get account/ticker info")
Sleep(5000)
continue
current_price = ticker['Last']
Log(f"[Price] 原始: {df['Close'].values[-1]:.2f}, "
f"平滑当前: {w_close_current:.2f}, 平滑前值: {w_close_prev:.2f}")
Log(f"[Trend] {'↑ 向上' if signal == 1 else '↓ 向下' if signal == -1 else '→ 横盘'}")
# 执行交易逻辑
if signal == 1 and position != 1:
# 平滑价格向上 → 做多
Log(f"[信号] 趋势向上，开多 @ {current_price:.2f}")
if position == -1:
# 先平空仓
exchange.SetDirection("closesell")
exchange.Buy(current_price, 1)
Log(f"[平仓] 平空仓")
# 开多仓
exchange.SetDirection("buy")
exchange.Buy(current_price, 1)
Log(f"[开仓] 开多仓")
position = 1
elif signal == -1 and position != -1:
# 平滑价格向下 → 做空
Log(f"[信号] 趋势向下，开空 @ {current_price:.2f}")
if position == 1:
# 先平多仓
exchange.SetDirection("closebuy")
exchange.Sell(current_price, 1)
Log(f"[平仓] 平多仓")
# 开空仓
exchange.SetDirection("sell")
exchange.Sell(current_price, 1)
Log(f"[开仓] 开空仓")
position = -1
else:
Log(f"[持仓] 当前{'多头' if position == 1 else '空头' if position == -1 else '空仓'}，无需操作")
もちろん、実際にはそれほど単純ではありません。レベル 2 で短期トレンド、レベル 4 で長期トレンドを示すなど、複数のレベルのウェーブレットを同時に使用できます。両方が同じ方向に動いている場合にのみポジションを開くことで、誤ったシグナルが大幅に減少します。また、取引量の増加、十分に高いボラティリティ、重要なレベルを超える価格ブレイクアウトなど、他のフィルタリング条件を追加することもできます。これらはすべて勝率を向上させることができます。ストップロス注文については、ウェーブレット平滑化価格変動範囲を使用して動的に設定できます。たとえば、価格が平滑化価格から ATR の 2 倍を引いた値を下回った場合にストップロスを設定します。ポジション管理では、トレンドが明らかなほど（平滑化価格の傾きが急なほど）、ポジション サイズが大きくなります。トレンドが不明確な場合は、ポジションを小さくするか、傍観者のままにします。
しかし、基本的な考え方は変わりません。ウェーブレットを用いてノイズの多い価格を明確なトレンドに変換し、その明確なトレンドに基づいて判断を下すのです。これは、元のローソク足チャートの上昇と下降を直接見るよりもはるかに信頼性が高いです。なぜなら、元のローソク足チャートは今日3%上昇し、明日2%下落し、その次の日は4%上昇する可能性があるからです。トレンドなのか変動なのかを区別することはできません。ウェーブレット処理された曲線は、「この期間の全体的なトレンドは上昇傾向にあるが、その間に変動がある」と示してくれます。
VIII. 結論：「三本柱のアプローチ」の合理的見解
実用的な平滑化効果という点では、ウェーブレット変換は金融データ処理において確かに役割を果たします。短期的なノイズを除去し、比較的明確なトレンド情報を抽出するのに役立つからです。しかし、この手法には重大な限界もあります。**ラグの問題は完全に回避することはできません。**しかし、ウェーブレット変換は過去のデータしか処理できず、将来のトレンドを予測することはできません。さらに、ウェーブレット変換のみを用いた場合の効果は比較的限定的であるため、完全な取引システムを構築するには、他の分析手法やリスク管理策と組み合わせる必要があります。
この限界の根本的な理由は、金融市場の特異性にあります。音声認識や画像処理といった従来の信号処理分野では、ノイズ特性は比較的安定しており、信号パターンは繰り返される傾向があります。そのため、ウェーブレット変換は信号とノイズを効果的に分離できます。しかし、金融市場は全く異なります。今日「ノイズ」とみなされる変動が、明日には市場の変化を反映する「シグナル」になる可能性があり、現在有効な分析モデルが将来的には有効でなくなる可能性があります。**市場自体は非定常であり、動的に変化します。**ウェーブレット変換には不変の法則がないため、金融分野におけるウェーブレット変換の適用は、特定の市場環境に応じて柔軟に調整する必要があります。
ウェーブレット変換やフーリエ変換の実際の効果を誇張している人を見かけたら、次のような質問をしてみてください。どのウェーブレットタイプが使用されましたか？他のタイプではなく、このタイプを選択した根拠は何ですか？平滑化レベルはどのように設定されましたか？対応するバックテスト結果やパラメータ選択手順はありますか？本当に専門知識を持つ人であれば、これらの重要な技術的詳細を明確に説明できるはずです。
私たちは限られた知識に基づいて、この実践的な調査を実施しました。**中心となるアイデアは、ウェーブレット変換の応用概念をシンプルでわかりやすい方法で共有することです。**この記事は、読者の皆様にこの技術の基礎的な理解を深めていただくことを目的としています。私たちは、この分野に深く携わる定量的研究者たちを高く評価しています。この分野の専門家の皆様には、ウェーブレットパラメータ選択の理論的根拠、マルチスケール組み合わせの最適化手法、適応型ウェーブレット選択の実装パスなど、本記事の不備点をご指摘いただければ幸いです。皆様からのご提案を真摯に受け止め、継続的に内容を改善してまいります。
プロット機能: 発明者のローカルバックテストエンジンに適用
python
'''backtest
start: 2025-12-17 00:00:00
end: 2025-12-23 08:00:00
period: 1h
basePeriod: 1h
exchanges: [{"eid":"Futures_Binance","currency":"BTC_USDT","fee":[0,0]}]
'''
from fmz import *
import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
from matplotlib.patches import Rectangle
task = VCtx(__doc__)
# ==================== 小波系数库 ====================
class WaveletCoefficients:
"""Wavelet Coefficients Definition"""
@staticmethod
def get_coeffs(wavelet_name):
"""Get coefficients for different wavelet types"""
coeffs = {
"Haar": [0.5, 0.5],
"Daubechies 4": [
0.48296291314453414,
0.8365163037378079,
0.22414386804201339,
-0.12940952255126037
],
"Symlet 4": [
-0.05357, -0.02096, 0.35238,
0.56833, 0.21062, -0.07007,
-0.01941, 0.03268
],
"Biorthogonal 3.3": [
-0.06629, 0.28289, 0.63678,
0.28289, -0.06629
],
"Mexican Hat (Ricker)": [
-0.1, 0.0, 0.4, 0.8, 0.4, 0.0, -0.1
],
"Morlet (Gaussian)": [
0.0625, 0.25, 0.375, 0.25, 0.0625
],
"Discrete Meyer (Dmey)": [
-0.015, -0.025, 0.0,
0.28, 0.52, 0.28,
0.0, -0.025, -0.015
]
}
return coeffs.get(wavelet_name, coeffs["Mexican Hat (Ricker)"])
# ==================== 小波变换引擎 ====================
class WaveletTransform:
"""Wavelet Transform Engine"""
def __init__(self, wavelet_type="Mexican Hat (Ricker)", smoothing_level=3):
self.wavelet_type = wavelet_type
self.smoothing_level = smoothing_level
self.coeffs = WaveletCoefficients.get_coeffs(wavelet_type)
def convolve(self, src, coeffs, step):
"""
Convolution operation - Core algorithm
Args:
src: Source data sequence
coeffs: Wavelet coefficients
step: Sampling step
Returns:
Convolved value
"""
sum_val = 0.0
sum_w = 0.0
for i, weight in enumerate(coeffs):
idx = i * step
if idx < len(src):
val = src[idx]
sum_val += val * weight
sum_w += weight
# Normalization - Critical fix
return sum_val / sum_w if sum_w != 0 else sum_val
def calc_level(self, data, target_level):
"""
Calculate wavelet transform for specified level
Args:
data: Original data array
target_level: Target smoothing level
Returns:
Transformed data array
"""
result = []
coeffs = self.coeffs
for i in range(len(data)):
# Get data from current position backwards
src = data[max(0, i - 50):i + 1][::-1]
# Level 1
val = self.convolve(src, coeffs, 1)
# Level 2
if target_level >= 2:
src_temp = [val] + [self.convolve(data[max(0, j - 50):j + 1][::-1], coeffs, 1)
for j in range(max(0, i - 10), i)][::-1]
val = self.convolve(src_temp, coeffs, 2)
# Level 3
if target_level >= 3:
val = self.convolve([val] * len(coeffs), coeffs, 4)
# Level 4+
if target_level >= 4:
val = self.convolve([val] * len(coeffs), coeffs, 8)
result.append(val)
return np.array(result)
def transform_ohlc(self, df):
"""
Perform wavelet transform on OHLC data
Args:
df: DataFrame with Open/High/Low/Close
Returns:
Transformed DataFrame
"""
result_df = df.copy()
# Transform each price series
result_df['w_open'] = self.calc_level(df['Open'].values, self.smoothing_level)
result_df['w_high'] = self.calc_level(df['High'].values, self.smoothing_level)
result_df['w_low'] = self.calc_level(df['Low'].values, self.smoothing_level)
result_df['w_close'] = self.calc_level(df['Close'].values, self.smoothing_level)
# Reconstruct logically consistent candlesticks
result_df['real_high'] = result_df[['w_high', 'w_low', 'w_open', 'w_close']].max(axis=1)
result_df['real_low'] = result_df[['w_high', 'w_low', 'w_open', 'w_close']].min(axis=1)
return result_df
# ==================== K线图可视化工具 ====================
class WaveletCandlestickVisualizer:
"""Wavelet Candlestick Visualization"""
@staticmethod
def plot_single_wavelet(df, wavelet_type, smoothing_level=3, n_bars=200):
"""
Plot single wavelet type comparison
Args:
df: Original candlestick data
wavelet_type: Wavelet type
smoothing_level: Smoothing level
n_bars: Number of bars to display
"""
# Take only last n_bars
df_plot = df.iloc[-n_bars:].copy()
# Create figure
fig, ax = plt.subplots(figsize=(20, 8))
# Perform wavelet transform
transformer = WaveletTransform(wavelet_type, smoothing_level)
transformed = transformer.transform_ohlc(df)
transformed_plot = transformed.iloc[-n_bars:].copy()
# Draw original candlesticks (gray background)
WaveletCandlestickVisualizer._draw_candlesticks(
ax, df_plot,
color_up='lightgray',
color_down='lightgray',
alpha=0.3,
label='Original Candles'
)
# Draw wavelet smoothed candlesticks
WaveletCandlestickVisualizer._draw_candlesticks(
ax, transformed_plot,
use_wavelet=True,
color_up='#26A69A', # Green
color_down='#EF5350', # Red
alpha=0.9,
linewidth=1.2,
label=f'{wavelet_type} Smoothed (Level {smoothing_level})'
)
# Set title and labels
ax.set_title(f'{wavelet_type} Wavelet (Level {smoothing_level}) - Candlestick Comparison',
fontsize=16, fontweight='bold', pad=20)
ax.set_ylabel('Price (USDT)', fontsize=13)
ax.set_xlabel('Time', fontsize=13)
ax.grid(True, alpha=0.2, linestyle='--')
ax.legend(loc='upper left', fontsize=12)
# Format x-axis
ax.set_xlim(-1, len(df_plot))
ax.set_xticks(range(0, len(df_plot), max(1, len(df_plot) // 10)))
ax.set_xticklabels([df_plot.index[i].strftime('%m-%d %H:%M')
for i in range(0, len(df_plot), max(1, len(df_plot) // 10))],
rotation=45, ha='right')
plt.tight_layout()
plt.show()
return fig
@staticmethod
def plot_single_level(df, wavelet_type, level, n_bars=200):
"""
Plot single smoothing level
Args:
df: Original candlestick data
wavelet_type: Wavelet type
level: Smoothing level
n_bars: Number of bars to display
"""
# Take only last n_bars
df_plot = df.iloc[-n_bars:].copy()
# Create figure
fig, ax = plt.subplots(figsize=(20, 8))
# Perform wavelet transform
transformer = WaveletTransform(wavelet_type, level)
transformed = transformer.transform_ohlc(df)
transformed_plot = transformed.iloc[-n_bars:].copy()
# Draw original candlesticks
WaveletCandlestickVisualizer._draw_candlesticks(
ax, df_plot,
color_up='lightgray',
color_down='lightgray',
alpha=0.3,
label='Original Candles'
)
# Draw wavelet smoothed candlesticks
WaveletCandlestickVisualizer._draw_candlesticks(
ax, transformed_plot,
use_wavelet=True,
color_up='#26A69A',
color_down='#EF5350',
alpha=0.9,
linewidth=1.2,
label=f'Level {level} Smoothed'
)
# Set title and labels
ax.set_title(f'{wavelet_type} - Smoothing Level {level} Effect',
fontsize=16, fontweight='bold', pad=20)
ax.set_ylabel('Price (USDT)', fontsize=13)
ax.set_xlabel('Time', fontsize=13)
ax.grid(True, alpha=0.2, linestyle='--')
ax.legend(loc='upper left', fontsize=12)
# Format x-axis
ax.set_xlim(-1, len(df_plot))
ax.set_xticks(range(0, len(df_plot), max(1, len(df_plot) // 10)))
ax.set_xticklabels([df_plot.index[i].strftime('%m-%d %H:%M')
for i in range(0, len(df_plot), max(1, len(df_plot) // 10))],
rotation=45, ha='right')
plt.tight_layout()
plt.show()
return fig
@staticmethod
def _draw_candlesticks(ax, df, use_wavelet=False, color_up='green',
color_down='red', alpha=1.0, linewidth=1.0, label=''):
"""
Draw candlestick chart
Args:
ax: Matplotlib axis
df: Data DataFrame
use_wavelet: Whether to use wavelet data
color_up: Up color
color_down: Down color
alpha: Transparency
linewidth: Line width
label: Legend label
"""
if use_wavelet:
opens = df['w_open'].values
highs = df['real_high'].values
lows = df['real_low'].values
closes = df['w_close'].values
else:
opens = df['Open'].values
highs = df['High'].values
lows = df['Low'].values
closes = df['Close'].values
for i in range(len(df)):
x = i
open_price = opens[i]
high_price = highs[i]
low_price = lows[i]
close_price = closes[i]
color = color_up if close_price >= open_price else color_down
# Draw wick
ax.plot([x, x], [low_price, high_price],
color=color, linewidth=linewidth, alpha=alpha)
# Draw body
height = abs(close_price - open_price)
bottom = min(open_price, close_price)
rect = Rectangle((x - 0.3, bottom), 0.6, height,
facecolor=color, edgecolor=color,
alpha=alpha, linewidth=linewidth)
ax.add_patch(rect)
# Add legend (only once)
if label:
ax.plot([], [], color=color_up, linewidth=3, alpha=alpha, label=label)
# ==================== 主函数 ====================
def main():
"""Main execution flow"""
exchange.SetCurrency("BTC_USDT")
exchange.SetContractType("swap")
# Get candlestick data
records = exchange.GetRecords(PERIOD_H1, 500)
# Convert to DataFrame
df = pd.DataFrame(records, columns=['Time', 'Open', 'High', 'Low', 'Close', 'Volume'])
df['Time'] = pd.to_datetime(df['Time'], unit='ms')
df.set_index('Time', inplace=True)
print(f"Data loaded: {len(df)} bars")
print(f"Time range: {df.index[0]} to {df.index[-1]}")
print(f"Price range: ${df['Low'].min():.2f} - ${df['High'].max():.2f}")
return df
# ==================== 执行绘图 ====================
try:
# Get candlestick data
kline = main()
print("\n" + "="*70)
print("Generating Wavelet Candlestick Charts (Each in Separate Window)...")
print("="*70)
# ========== Chart Series 1: Different Wavelet Types ==========
print("\n[Series 1] Comparing Different Wavelet Types")
print("-" * 70)
wavelet_types = [
"Haar",
"Daubechies 4",
"Symlet 4",
"Biorthogonal 3.3",
"Mexican Hat (Ricker)",
"Morlet (Gaussian)",
"Discrete Meyer (Dmey)" # ✅ 添加了 Discrete Meyer
]
for i, wavelet_type in enumerate(wavelet_types, 1):
print(f" Chart {i}/{len(wavelet_types)}: {wavelet_type}")
fig = WaveletCandlestickVisualizer.plot_single_wavelet(
kline,
wavelet_type=wavelet_type,
smoothing_level=3,
n_bars=150
)
# ========== Chart Series 2: Different Smoothing Levels ==========
print("\n[Series 2] Comparing Different Smoothing Levels")
print("-" * 70)
levels = [1, 2, 3, 4, 5]
for i, level in enumerate(levels, 1):
print(f" Chart {i}/5: Level {level}")
fig = WaveletCandlestickVisualizer.plot_single_level(
kline,
wavelet_type="Mexican Hat (Ricker)",
level=level,
n_bars=150
)
print("\n" + "="*70)
print("All charts generated successfully!")
print(f"Total charts: {len(wavelet_types) + len(levels)} ({len(wavelet_types)} wavelets + {len(levels)} levels)")
print("="*70)
except Exception as e:
print(f"Error: {str(e)}")
import traceback
print(traceback.format_exc())
finally:
print("\nStrategy testing completed.")
トランザクション機能: Inventorsプラットフォームに適用
python
'''backtest
start: 2025-01-17 00:00:00
end: 2025-12-23 08:00:00
period: 1h
basePeriod: 1h
exchanges: [{"eid":"Futures_Binance","currency":"BTC_USDT","fee":[0,0]}]
'''
import numpy as np
import pandas as pd
# ==================== 小波系数库 ====================
class WaveletCoefficients:
"""与上部分函数一致"""
# ==================== 小波变换引擎 ====================
class WaveletTransform:
"""与上部分函数一致"""
def main():
"""小波交易主函数 - 基于平滑价格趋势"""
# ========== 配置参数 ==========
WAVELET_TYPE = "Mexican Hat (Ricker)" # 小波类型
SMOOTHING_LEVEL = 1 # 平滑阶数
# 初始化
exchange.SetCurrency("BTC_USDT")
exchange.SetContractType("swap")
Log(f"=" * 70)
Log(f"Wavelet Trend Following Strategy")
Log(f"Wavelet: {WAVELET_TYPE}, Level: {SMOOTHING_LEVEL}")
Log(f"Logic: 平滑收盘价向上→做多, 平滑收盘价向下→做空")
Log(f"=" * 70)
# 初始化小波变换器
transformer = WaveletTransform(WAVELET_TYPE, SMOOTHING_LEVEL)
# 持仓状态
position = 0 # 0: 无持仓, 1: 多头, -1: 空头
while True:
# 获取K线数据
records = exchange.GetRecords(PERIOD_H1, 500)
if not records:
Log("[Warning] Failed to get kline data")
Sleep(5000)
continue
df = pd.DataFrame(records, columns=['Time', 'Open', 'High', 'Low', 'Close', 'Volume'])
df['Time'] = pd.to_datetime(df['Time'], unit='ms')
df.set_index('Time', inplace=True)
# 执行小波变换
transformed = transformer.transform_ohlc(df)
# 获取最近两根K线的平滑收盘价
w_close_current = transformed['w_close'].values[-1] # 当前平滑收盘价
w_close_prev = transformed['w_close'].values[-2] # 前一根平滑收盘价
# 判断趋势方向
signal = 0
if w_close_current > w_close_prev:
signal = 1 # 平滑价格向上 → 做多
elif w_close_current < w_close_prev:
signal = -1 # 平滑价格向下 → 做空
# 获取账户信息
account = exchange.GetAccount()
ticker = exchange.GetTicker()
if not account or not ticker:
Log("[Warning] Failed to get account/ticker info")
Sleep(5000)
continue
current_price = ticker['Last']
Log(f"[Price] 原始: {df['Close'].values[-1]:.2f}, "
f"平滑当前: {w_close_current:.2f}, 平滑前值: {w_close_prev:.2f}")
Log(f"[Trend] {'↑ 向上' if signal == 1 else '↓ 向下' if signal == -1 else '→ 横盘'}")
# 执行交易逻辑
if signal == 1 and position != 1:
# 平滑价格向上 → 做多
Log(f"[信号] 趋势向上，开多 @ {current_price:.2f}")
if position == -1:
# 先平空仓
exchange.SetDirection("closesell")
exchange.Buy(current_price, 1)
Log(f"[平仓] 平空仓")
# 开多仓
exchange.SetDirection("buy")
exchange.Buy(current_price, 1)
Log(f"[开仓] 开多仓")
position = 1
elif signal == -1 and position != -1:
# 平滑价格向下 → 做空
Log(f"[信号] 趋势向下，开空 @ {current_price:.2f}")
if position == 1:
# 先平多仓
exchange.SetDirection("closebuy")
exchange.Sell(current_price, 1)
Log(f"[平仓] 平多仓")
# 开空仓
exchange.SetDirection("sell")
exchange.Sell(current_price, 1)
Log(f"[开仓] 开空仓")
position = -1
else:
Log(f"[持仓] 当前{'多头' if position == 1 else '空头' if position == -1 else '空仓'}，无需操作")
Log(f"[账户] 余额: {account['Balance']:.2f}, 权益: {account['Equity']:.2f}")
Log("-" * 70)
Sleep(60000 * 60)
- 1