Table of Contents
The concept of the Moving Average (MA) originates from the Dow Theory developed in the early 20th century. Dow Theory emphasizes that markets exhibit trends, and such trends can be observed through price movements themselves. However, in the early days, price data was highly volatile and lacked smoothing tools. Analysts therefore began using the average price over a certain period to reduce random noise—this marked the beginning of the moving average.
With advancements in statistics and computer computation, the moving average gradually became one of the most fundamental technical indicators in quantitative trading. From the Simple Moving Average (SMA) and Exponential Moving Average (EMA) to more sophisticated variants such as the Double Exponential Moving Average (DEMA) and Hull Moving Average (HMA), all these improvements aim to address the inherent lag problem of traditional moving averages, allowing them to reflect price trends more quickly or more smoothly.
In practice, moving average strategies are widely used across different markets:
Among these, the most representative examples are the Golden Cross and Death Cross. When a short-term moving average crosses above a long-term moving average, it is often regarded as the beginning of a bullish trend; conversely, when it crosses below, it signals a potential bearish trend. However, relying solely on crossover signals in a sideways or choppy market can lead to false breakouts. Therefore, in practice, traders often incorporate layered stop-loss mechanisms, volatility filters, and futures rollover adjustments to improve the strategy’s feasibility and stability.
This strategy, built within the Golden Cross framework, integrates trailing stop-losses and dynamic filtering for risk management, and applies continuous contract data to handle futures rollovers, ensuring consistency between backtesting and live trading performance.
# Trading Flowchart
At each daily close → Update filter status
├─ Is the filter OFF?
│ └─ No (Filter = ON)
│ ├─ Have position → Force close (start 5-trading-day cooldown)
│ └─ No position → No entry
└─ Yes (Filter = OFF)
├─ Have position
│ ├─ Stop-loss (1D/5D/10D) → Close (start 5-day cooldown)
│ ├─ Death Cross → Close
│ ├─ Holding > 200 days → Close
│ └─ Otherwise → Continue holding
└─ No position
├─ (If in cooldown) → No entry; cooldown −1
├─ Golden Cross (with 1.0001 buffer) → Enter full position (leverage 1.8×)
└─ Else → No entry
#%% Setup
ticker1 = 'IR0001 IX0001'
ticker2 = 'MTX MSCI NYF'
# 環境變數
import os
import sys
# import time # 未使用
import yaml
''' ------------------- 不使用 config.yaml 管理 API KEY 的使用者可以忽略以下程式碼 -------------------'''
notebook_dir = os.path.dirname(os.path.abspath(__file__)) if '__file__' in globals() else os.getcwd()
yaml_path = os.path.join(notebook_dir, '..', 'config.yaml')
yaml_path = os.path.abspath(os.path.join(notebook_dir, '..', 'config.yaml'))
with open(yaml_path, 'r') as tejapi_settings: config = yaml.safe_load(tejapi_settings)
''' ------------------- 不使用 config.yaml 管理 API KEY 的使用者可以忽略以上程式碼 -------------------'''
# --------------------------------------------------------------------------------------------------
os.environ['TEJAPI_BASE'] = config['TEJAPI_BASE'] # = "https://api.tej.com.tw"
os.environ['TEJAPI_KEY'] = config['TEJAPI_KEY'] # = "YOUR_API_KEY"
# --------------------------------------------------------------------------------------------------
os.environ['ticker'] = ticker1
os.environ['future'] = ticker2
os.environ['mdate'] = '20180101 20251002'
!zipline ingest -b tquant_future
# 數據分析套件
import warnings
import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
from logbook import Logger, StderrHandler, INFO
# tquant 相關套件
# import tejapi
# import TejToolAPI
import zipline
import pyfolio as pf
from zipline.utils.calendar_utils import get_calendar
from zipline.utils.events import date_rules, time_rules
from zipline.finance.commission import (
PerContract
)
from zipline.finance.slippage import (
FixedSlippage
)
from zipline import run_algorithm
from zipline.api import (
record,
schedule_function,
set_slippage,
set_commission,
order_value,
set_benchmark,
symbol,
get_datetime,
date_rules,
time_rules,
continuous_future
)
from pyfolio.utils import extract_rets_pos_txn_from_zipline
# logbook 設定
warnings.filterwarnings('ignore')
print(sys.executable)
print(sys.version)
print(sys.prefix)
log_handler = StderrHandler(
format_string = (
'[{record.time:%Y-%m-%d %H:%M:%S.%f}]: ' +
'{record.level_name}: {record.func_name}: {record.message}'
),
level=INFO
)
log_handler.push_application()
log = Logger('Algorithm')
#%% Strategy 1.3
# 目前最讚的版本的修改版 + 濾網天數限制
'''========================================================================================='''
# 策略參數
#---------------------------------------------------------------------------------------------
'''
未觸發濾網時 MA3 & MA10 黃金交叉 => 買進(全倉，槓桿 1.8倍)，濾網開啟 or 死亡交叉 or 觸發停損條件 or 超過持有天數 => 平倉
'''
# 均線參數
SHORT_MA = 3
LONG_MA = 10
# 停損參數
MAX_HOLD_DAYS = 200
STOPLOSS_1D = 5
STOPLOSS_5D = 10
STOPLOSS_10D = 15
# 濾網參數
FILLTER_ON_ACCDAY = 2
FILLTER_OF_ACCDAY = 3
FILLTER_ON_TRIGGER = -5
FILLTER_OF_TRIGGER = 5
FILLTER_AUTO_OFF_DAYS = 200 # 濾網開啟後自動關閉天數
# 槓桿參數
LEVERAGE = 1.8
MAX_LEVERAGE = LEVERAGE + 0.05
# 其他
MTX_CONTRACT_MULTIPLIER = 50 # 小台乘數（請依你行情系統設定）
CAPITAL_BASE = 1e6
START_DT = pd.Timestamp('2020-01-02', tz='utc')
END_DT = pd.Timestamp('2025-09-26', tz='utc')
print(f'最大槓桿：{MAX_LEVERAGE} 倍')
print("[Trading log---------------]: ----------------------------- | signal ----- |action|Original posi| Note")
#---------------------------------------------------------------------------------------------
def initialize(context):
set_benchmark(symbol('IR0001'))
set_commission(
futures=PerContract(
cost = {'MTX': 15},
exchange_fee = 0
)
)
set_slippage(
futures=FixedSlippage(spread=1.0))
# set_max_leverage(MAX_LEVERAGE)
context.asset = continuous_future('MTX', offset=0, roll='calendar', adjustment='add')
context.universe = [context.asset]
context.in_position = False
context.hold_days = 0
context.wait_after_stoploss = 0
context.filter_triggered = 0 # 添加濾網狀態變數
context.filter_days = 0 # 濾網開啟天數計數器
schedule_function(ma_strategy, date_rules.every_day(), time_rules.market_close())
def ma_strategy(context, data):
# 獲取足夠的歷史資料
hist = data.history(context.asset, ['close'], bar_count=max(LONG_MA+2, 15), frequency='1d')
close = hist['close']
# 獲取基準指標歷史資料
benchmark_hist = data.history(symbol('IR0001'), ['close'], bar_count=7, frequency='1d')
benchmark_close = benchmark_hist['close']
# 計算策略累計報酬
return_1d = (close.iloc[-1] - close.iloc[-2]) / close.iloc[-2] * 100
return_5d = (close.iloc[-1] - close.iloc[-6]) / close.iloc[-6] * 100
return_10d = (close.iloc[-1] - close.iloc[-11]) / close.iloc[-11] * 100
# 計算指數累計報酬
benchmark_return_on = (benchmark_close.iloc[-1] - benchmark_close.iloc[-(FILLTER_ON_ACCDAY+1)]) / benchmark_close.iloc[-(FILLTER_ON_ACCDAY+1)] * 100
benchmark_return_of = (benchmark_close.iloc[-1] - benchmark_close.iloc[-(FILLTER_OF_ACCDAY+1)]) / benchmark_close.iloc[-(FILLTER_OF_ACCDAY+1)] * 100
# 計算均線
short_ma = close.rolling(window=SHORT_MA).mean()
long_ma = close.rolling(window=LONG_MA).mean()
# 判斷是否有持倉
contract = data.current(context.asset, 'contract')
open_positions = context.portfolio.positions
in_position = contract in open_positions and open_positions[contract].amount != 0
value = context.portfolio.portfolio_value
# 計算黃金交叉訊號
# golden_cross = (short_ma.iloc[-2] < long_ma.iloc[-2]) and (short_ma.iloc[-1] > long_ma.iloc[-1])
golden_cross = (short_ma.iloc[-2] < long_ma.iloc[-2]) and (short_ma.iloc[-1] > long_ma.iloc[-1] * 1.0001)
# 計算死亡交叉訊號
death_cross = (short_ma.iloc[-2] > long_ma.iloc[-2]) and (short_ma.iloc[-1] < long_ma.iloc[-1])
# 計算停損訊號
stop_loss_flag = (return_1d < -STOPLOSS_1D) or (return_5d < -STOPLOSS_5D) or (return_10d < -STOPLOSS_10D)
# 計算濾網訊號
if context.filter_triggered == 0:
if benchmark_return_on <= FILLTER_ON_TRIGGER:
context.filter_triggered = 1
context.filter_days = 0 # 重置濾網天數計數器
#----------------------------------|--------------|
log.info(f'{get_datetime().date()} | Filter ON | Benchmark {FILLTER_ON_ACCDAY}d return: {benchmark_return_on:.2f}%')
else: # context.filter_triggered == 1
context.filter_days += 1 # 濾網天數計數器增加
# 檢查濾網自動關閉條件：天數達到上限
if context.filter_days >= FILLTER_AUTO_OFF_DAYS:
context.filter_triggered = 0
context.filter_days = 0
#----------------------------------|--------------|
log.info(f'{get_datetime().date()} | Filter OFF | Auto off after {FILLTER_AUTO_OFF_DAYS} days')
# 檢查濾網關閉條件：市場反彈
elif benchmark_return_of >= FILLTER_OF_TRIGGER:
context.filter_triggered = 0
context.filter_days = 0
#----------------------------------|--------------|
log.info(f'{get_datetime().date()} | Filter OFF | Benchmark {FILLTER_OF_ACCDAY}d return: {benchmark_return_of:.2f}%')
# 計算進出場訊號
# 濾網未觸發時，依策略進出場
if context.filter_triggered == 0:
# 有部位 => 判斷出場時機
if in_position:
context.hold_days += 1
# 移動停損
if stop_loss_flag:
order_value(contract, 0)
#----------------------------------|--------------|
log.info(f'{get_datetime().date()} | STOP_LOSS | sell | position: {open_positions[contract].amount if contract in open_positions else 0} | Hold Days: {context.hold_days}')
context.hold_days = 0
context.wait_after_stoploss = 5
# 死亡交叉
elif death_cross:
order_value(contract, 0)
#----------------------------------|--------------|
log.info(f'{get_datetime().date()} | Death Cross | sell | position: {open_positions[contract].amount if contract in open_positions else 0} | Hold Days: {context.hold_days}')
context.hold_days = 0
# 超過持有天數限制
elif context.hold_days >= MAX_HOLD_DAYS:
order_value(contract, 0)
#----------------------------------|--------------|
log.info(f'{get_datetime().date()} | Max Hold | sell | position: {open_positions[contract].amount if contract in open_positions else 0} | Hold Days: {context.hold_days}')
context.hold_days = 0
# 無部位 => 判斷進場時機
else:
context.hold_days = 0
if context.wait_after_stoploss > 0:
context.wait_after_stoploss -= 1
elif golden_cross:
order_value(contract, value * LEVERAGE)
#----------------------------------|--------------|
log.info(f'{get_datetime().date()} | Golden Cross | buy | position: {open_positions[contract].amount if contract in open_positions else 0} | Amount: {value * LEVERAGE}')
context.hold_days = 0
# 濾網觸發時，無條件直接平倉
else: # context.filter_triggered == 1
if in_position:
context.hold_days += 1
order_value(contract, 0)
#----------------------------------|--------------|
log.info(f'{get_datetime().date()} | Filter Close | sell | position: {open_positions[contract].amount if contract in open_positions else 0} | Hold Days: {context.hold_days} | Filter Days: {context.filter_days}')
context.hold_days = 0
context.wait_after_stoploss = 5
record(close=data.current(context.asset, 'price'))
record(position=open_positions[contract].amount if contract in open_positions else 0)
record(filter_status=context.filter_triggered) # 記錄濾網狀態
def analyze(context=None, results=None):
close = results['close']
ma3 = close.rolling(window=3).mean()
ma10 = close.rolling(window=10).mean()
position = results['position']
filter_status = results.get('filter_status', pd.Series(0, index=results.index)) # 濾網狀態
fig, (ax1, ax2, ax3) = plt.subplots(
3, 1,
figsize=(16, 9),
sharex=True,
gridspec_kw={'height_ratios': [4, 2, 2]}
)
# 上方：策略與基準績效
results.algorithm_period_return.plot(label='Strategy', ax=ax1)
results.benchmark_period_return.plot(label='Benchmark', ax=ax1)
ax1.set_title('MTX 3MA/10MA Crossover Strategy with Filter (Dynamic Contracts)')
ax1.legend(loc="upper left")
ax1.grid(True)
# 中間：收盤價、均線與持倉區塊
# ax2.plot(close.index, close, label='MTX Close')
ma_diff = ma3 - ma10
ax2.set_title('MTX Close, MA3, MA10 with Position Highlight')
ax2.bar(ma_diff.index, ma_diff, width=0.8, alpha=0.7, label='MA3-MA10',
color=['black' if x > 0 else 'black' for x in ma_diff])
ax2.axhline(y=0, color='black', linewidth=2, linestyle='-')
ax2.set_ylabel('MA3 - MA10')
ax2.legend(loc="upper left")
ax2.grid(True)
# ax2.plot(ma3.index, ma3, label='MA3')
# ax2.plot(ma10.index, ma10, label='MA10')
# 下方：濾網狀態
ax3.plot(filter_status.index, filter_status, label='Filter Status', color='red', linewidth=2)
ax3.fill_between(filter_status.index, 0, filter_status, alpha=0.3, color='red',
where=(filter_status > 0), label='Filter ON')
ax3.set_ylim(-0.1, 1.1)
ax3.set_ylabel('Filter Status')
ax3.set_title('Filter Status (0=OFF, 1=ON)')
ax3.legend(loc="upper left")
ax3.grid(True)
# 添加持倉區塊到所有圖表
in_position = (position > 0)
start = None
for i in range(len(in_position)):
if in_position.iloc[i] and start is None:
start = i
elif not in_position.iloc[i] and start is not None:
ax1.axvspan(close.index[start], close.index[i-1], color='orange', alpha=0.3)
ax2.axvspan(close.index[start], close.index[i-1], color='orange', alpha=0.3)
ax3.axvspan(close.index[start], close.index[i-1], color='orange', alpha=0.3)
start = None
if start is not None:
ax1.axvspan(close.index[start], close.index[-1], color='orange', alpha=0.3)
ax2.axvspan(close.index[start], close.index[-1], color='orange', alpha=0.3)
ax3.axvspan(close.index[start], close.index[-1], color='orange', alpha=0.3)
plt.tight_layout()
plt.show()
results = run_algorithm(
start = START_DT,
end = END_DT,
initialize = initialize,
capital_base = CAPITAL_BASE,
analyze = analyze,
data_frequency = 'daily',
bundle = 'tquant_future',
trading_calendar = get_calendar('TEJ'),
)
#%% pyfolio
'''========================================================================================='''
plt.rcParams['font.sans-serif'] = ['Arial', 'Noto Sans CJK TC', 'SimHei']
plt.rcParams['axes.unicode_minus'] = False
try:
returns, positions, transactions = extract_rets_pos_txn_from_zipline(results)
# print('returns:', returns.head())
# print('positions:', positions.head())
# print('transactions:', transactions.head())
except Exception as e:
print('extract_rets_pos_txn_from_zipline error:', e)
returns = results.get('algorithm_period_return', None)
positions = None
transactions = None
if returns is not None:
print('returns (fallback):', returns.head())
benchmark_rets = getattr(results, 'benchmark_return', None)
if benchmark_rets is None and hasattr(results, 'benchmark_period_return'):
benchmark_rets = results.benchmark_period_return
if benchmark_rets is not None:
print('benchmark_rets:', benchmark_rets.head())
else:
print('No benchmark returns found!')
if returns is not None:
pf.tears.create_full_tear_sheet(
returns=returns,
positions=positions,
transactions=transactions,
benchmark_rets=benchmark_rets
)
else:
print('No returns data for pyfolio.')
#%% Summary
'''========================================================================================='''
summary_strategy = pf.timeseries.perf_stats(returns, factor_returns=benchmark_rets)
summary_benchmark = pf.timeseries.perf_stats(benchmark_rets)
summary = pd.concat([summary_strategy, summary_benchmark], axis=1)
summary.columns = ['Strategy', 'Benchmark']
summary = summary.round(4)
summary
benchmark_rets: 2020-01-02 00:00:00+00:00 0.008614
2020-01-03 00:00:00+00:00 0.000823
2020-01-06 00:00:00+00:00 -0.012970
2020-01-07 00:00:00+00:00 -0.006110
2020-01-08 00:00:00+00:00 -0.005322
Name: benchmark_return, dtype: float64
| Start date | 2020-01-02 | |
|---|---|---|
| End date | 2025-09-26 | |
| Total months | 66 | |
| Backtest | ||
| Annual return | 29.162% | |
| Cumulative returns | 312.722% | |
| Annual volatility | 13.704% | |
| Sharpe ratio | 1.94 | |
| Calmar ratio | 2.67 | |
| Stability | 0.81 | |
| Max drawdown | -10.902% | |
| Omega ratio | 1.88 | |
| Sortino ratio | 3.28 | |
| Skew | 0.54 | |
| Kurtosis | 13.65 | |
| Tail ratio | 1.96 | |
| Daily value at risk | -1.621% | |
| Gross leverage | 0.61 | |
| Daily turnover | 0.156% | |
| Alpha | 0.24 | |
| Beta | 0.28 | |
| Worst drawdown periods | Net drawdown in % | Peak date | Valley date | Recovery date | Duration |
|---|---|---|---|---|---|
| 0 | 10.90 | 2024-07-11 | 2024-09-04 | 2025-05-09 | 198 |
| 1 | 10.30 | 2023-07-14 | 2023-08-25 | 2024-05-13 | 201 |
| 2 | 9.16 | 2021-04-20 | 2022-07-06 | 2022-07-20 | 309 |
| 3 | 6.69 | 2020-10-12 | 2020-10-30 | 2020-11-06 | 20 |
| 4 | 6.09 | 2023-03-07 | 2023-07-10 | 2023-07-14 | 88 |
| Top 10 long positions of all time | max |
|---|---|
| sid | |
| MTX202009 | 64.14% |
| MTX202007 | 64.14% |
| MTX202401 | 63.85% |
| MTX202011 | 63.68% |
| MTX202309 | 63.38% |
| MTX202308 | 63.30% |
| MTX202101 | 63.26% |
| MTX202010 | 63.23% |
| MTX202207 | 63.08% |
| MTX202312 | 63.05% |
| Top 10 short positions of all time | max |
|---|---|
| sid |
| Top 10 positions of all time | max |
|---|---|
| sid | |
| MTX202009 | 64.14% |
| MTX202007 | 64.14% |
| MTX202401 | 63.85% |
| MTX202011 | 63.68% |
| MTX202309 | 63.38% |
| MTX202308 | 63.30% |
| MTX202101 | 63.26% |
| MTX202010 | 63.23% |
| MTX202207 | 63.08% |
| MTX202312 | 63.05% |
| Strategy | Benchmark | |
|---|---|---|
| Annual return | 0.2916 | 0.1881 |
| Cumulative returns | 3.1272 | 1.5977 |
| Annual volatility | 0.1370 | 0.1978 |
| Sharpe ratio | 1.9365 | 0.9707 |
| Calmar ratio | 2.6749 | 0.6587 |
| Stability | 0.8124 | 0.8224 |
| Max drawdown | -0.1090 | -0.2855 |
| Omega ratio | 1.8757 | 1.1919 |
| Sortino ratio | 3.2789 | 1.3591 |
| Skew | 0.5411 | -0.5698 |
| Kurtosis | 13.6454 | 8.1531 |
| Tail ratio | 1.9604 | 1.0142 |
| Daily value at risk | -0.0162 | -0.0242 |
| Alpha | 0.2355 | NaN |
| Beta | 0.2803 | NaN |
This strategy is built around the MA3/MA10 Golden Cross, with a 1.0001 buffer added to reduce false breakouts. It incorporates a sharp-decline filter (triggered when the benchmark’s 2-day drop ≤ −5%, enforcing immediate liquidation and a 5-day cooldown), a three-tier stop-loss system, and a time-based stop. The design allows the strategy to amplify gains during trending markets while prioritizing capital preservation during sharp drawdowns. A continuous contract is used to handle rollovers, improving consistency between backtesting and live trading results.
The advantages include clear trading signals, structured risk management, and a balanced 1.8× leverage that provides stable risk-adjusted returns under strict stop-loss control. Limitations include lag inherent in moving averages, vulnerability to whipsaws in sideways markets, and sensitivity of performance to filter thresholds and parameter tuning.
Future work should include walk-forward analysis and sensitivity testing (on MA periods, buffer values, and filter thresholds), integrating volatility-based regime segmentation or ADX filters to activate trading only during strong trends, and incorporating transaction cost and slippage evaluation to enhance real-world feasibility.
Note: This analysis is for informational purposes only and does not constitute investment advice or recommendations for any financial product.
【TQuant Lab Backtesting System】Solving Your Quantitative Finance Challenges
