【日內交易 (Day Trading) 的 Street smart 分析方法 (上篇)】
Added 2025-03-01 14:43:13 +0000 UTC今篇文講一個重點，點做 Day trade 量化研究 + 回測。(完整Codes我放左係篇文最底。另外團隊會儘快上載進階Alpha 第三段片，如果咁都提升唔到你地策略的sharpe，我諗我都無咩好再講了 : ))
事關好幾個月前已經有朋友叫我寫多啲 Intraday trading (亦即 day trade)，今篇文就同大家睇下行內的 Quant trader (並唔係Quant researcher，因為今日的分析可能聽落去會有啲discretionary同唔係非常quantitative robust，但我間唔中都用而個方法去同對面team discretionary trader交流個市，所以做做下發覺唔錯) 點樣去分析日內股價走勢。
我諗而度有試過做 Day trade 的朋友有好多，但通常做得 day trade 的朋友係岩岩開始trade既時侯都唔會話點樣賺到大錢或者穩定賺到錢，比較多的是少賺然後突然爆倉/補倉，或者穩定地蝕錢。原因其實都係不外乎一樣事，就係做day trade因為交易數量比較多，係無特別交易優勢的情況下，你的交易成績就會比interday trading 或 buy and hold investor 更快地反映出一個交易者的實際交易能力，而個就係大數法則的生活體驗。
所以，啲人話做 day trade 好易賺錢，新人一入黎就應該先 day trade，咁大家就應該先諗諗講而句野既人有咩目的，因為行內真係做trading的人都知道，day trade 係技術含量好高的一個交易頻率，特別係 systematic auto trading。
講返正題，今次我會用 LLY.US 的一分鐘數據去做分析，主要俾大家了解下最基礎啲行家點做分析同點樣得出大概預測。我會先由數據處理開始講，然後用一啲 codes result 去講解個分析優勢係邊，用的方法不外乎都係 regression。Regression 即係有一堆數據，例如一堆 x-axis 係前30分鐘回報 及 y-axis 係黎緊 30分鐘的回報，然後要你畫一條直線去best fit 而堆點，咁而條直線就係你用黎預測下一個時間段的一個benchmark。
直接開波，先睇下下面 LLY.US 的一分鐘數據 (無adjust):
Format 格式就係 datetime, open, high, low, close, volume。用的時間timezone係eastern time zone (即冬令時侯比香港慢 13 hours; 夏令慢 12hours)。
但你會發現兩個問題要先處理:
1. 通常我地只做09:30 - 16:00 交易，所以 outside Regular Trading Hour (RTH) 我地係唔要個啲數據， e.g. 08:00 要filter 走
2. 有啲1分鐘數據有缺失，假設data quality 上無問題，咁代表個一分鐘真係無成交發生，咁通常就會補返個 natural 1 minute time interval 落去，令到個數據睇落 smooth 啲同埋日後有需要同其他數據分拼個陣方便啲。
但你會發覺，當你補返啲natural 1 minute time intervals 個陣，而啲OHLCV會變晒 nan。而個時侯，通常做法係:
For OHLC:
Step1. Fill forward with previous close
Step2. Fill backward with previous close
點解要咁樣Fill，原因係係現實世界中，如果係無trade，即係之前有trade occurred 的 close係果個沒有交易的一分鐘仍然係而個close，Fill forward 可以做到而個效果。如果你fill forward with columns 自己而唔係之前的close，就會造成data errors， 因為上一分鐘的high 非常unlikely 會係你今一分鐘的 high，因為今一分鐘的high如果無volume咁只會係上一分鐘的close，除非上一分鐘的close = 上一分鐘的high。
至於 fill backward 因為我的回測只用到有正數volume的行數，而本身需要fill backward 的原因就正正係因為個d行數本身都無trade無volume，所以fill backward 的OHLC不影響我的回測，我只係唔想nan出現而導致run codes有機會出error。但back唔backfill，睇大家需要，有啲朋友驚失真可以唔back fill。
至於 V:
Simply fill with zero，反映現實。
—------------------------------------------------------------------------------------------------------------------------
因為今次係第一次同大家睇點處理intraday trading，所以我update埋python codes:
先載入libraries:
import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
import seaborn as sns
from datetime import datetime, time
import mpld3
from mpld3 import plugins
提一提醒，mpld3 唔少人應該都未裝，可以用 pip install mpld3 裝就可以。
然後整一個處理以上data preprocessing的code block:
def load_and_process_data(file_path, ticker):
"""
Load and process 1-minute stock data from txt file.
"""
# Read data from txt file
df = pd.read_csv(file_path, header=None,
names=['datetime', 'open', 'high', 'low', 'close', 'volume'])
# Convert datetime to proper format
df['datetime'] = pd.to_datetime(df['datetime'])
df.set_index('datetime', inplace=True)
# Filter for trading hours (9:30 to 16:00)
df = df.between_time('09:30', '16:00')
# Create a complete range of 1-minute timestamps
trading_days = np.unique(df.index.date)
all_minutes = []
for day in trading_days:
trading_start = pd.Timestamp.combine(day, time(9, 30))
trading_end = pd.Timestamp.combine(day, time(16, 0))
day_minutes = pd.date_range(start=trading_start, end=trading_end, freq='1min')
all_minutes.extend(day_minutes)
# Create new DataFrame with all minute intervals
full_df = pd.DataFrame(index=pd.DatetimeIndex(all_minutes))
# Join the actual data
full_df = full_df.join(df)
# First fill close with forward fill and backward fill
full_df['close'] = full_df['close'].fillna(method='ffill').fillna(method='bfill')
# Now use close to fill missing values in other OHLC columns
for col in ['open', 'high', 'low']:
# First try to forward fill values that exist
full_df[col] = full_df[col].fillna(method='ffill')
# Then fill any remaining NAs with the close price
full_df[col] = full_df[col].fillna(full_df['close'])
# Fill NA with 0 for volume
full_df['volume'] = full_df['volume'].fillna(0)
# Add ticker column
full_df['ticker'] = ticker
return full_df
到而步，我講講其中一個行內普遍做法，就係將 前30分鐘的 returns regress on 後30分鐘的 returns，睇下有無咩特別發現。咁係做之前，當然要將 1分鐘數據resample到30分鐘。結果如下:
Resampling 的 codes以下:
def resample_and_calculate_returns(df):
"""
Resample 1-minute data to 30-minute intervals and calculate returns.
"""
# Resample to 30-minute data
resampled = df.resample('30min').agg({
'open': 'first',
'high': 'max',
'low': 'min',
'close': 'last',
'volume': 'sum',
'ticker': 'first'
})
# Filter again to ensure we only have trading hours (9:30 to 16:00)
resampled = resampled.between_time('09:30', '16:00')
# Calculate returns
resampled['returns'] = resampled['close'].pct_change()
return resampled
Resampling完，我地先將前30分鐘的 returns regress on 後30分鐘的 returns。而度用到200日的一分鐘數據，結果如下:
而度大家一定要記得，個 regression (畫線) 其實係做緊咩。佢唔係就咁用一條regression apply on all the data。個code係先將最近200日的09:30 - 10:00 的回報拿出來，再regress 去 最近200日的 10:00 - 10:30 的回報數據，而點好重要，務必記得。
另外，你會睇到，絕大部份的regression都只顯示非常低的correlation，除左14:30-15:00 預測 15:00-15:30的有約-0.285的效果。而度我分享個小秘密俾大家知，你有無發覺，即使用左 regression，但好似都得唔到太大可以直接用上手的交易概念，因為 regression 就咁睇個 dependent variable returns 並唔高，就算估中expected returns，都係賺30分鐘之後的0 - 0.7% (in this example)的回報，而且隨時間 regression line’s slope 好大機會會變，變左今日的分析可能幾十天後就用唔著。
但個關鍵點唔係賺個0-0.7%，而係如果你有策略要係15:00-15:30入市而你要搵個合適入市位，好多時你都未必會有頭緒。到底係15:00 open入? 定點樣入?
我同班discretionary trader 俾既意見係，假設佢地有人想係 15:00-15:30 入市，唔好睇咩 R^2 咩 Beta 咩野，佢地靜係需要知道，按最近200日去睇，如果14:30-15:00 係升 (例如升左0.2%)，佢15:00-15:30 的回報應該大機會係會微跌，但跌幾多，無人知。但係我知既係，睇返幅圖，如果之前升左0.2%，最差15:00-15:30既時侯係會跌0.3%左右，所以如果係15:00-15:30期間有跌近0.3%，而個就係相對較好的買入位。
咁如果一開波就升左上去呢? 咁就調返轉諗升到幾多佢大機會調頭再入手，道理一樣。而個係for一啲本身係某段時間要入市但又唔知點pick位好，大家可以試下 forward testing，對我同一眾交易員黎講，好多時間比睇支持阻力更好。
大家今篇文先學習點用intraday regression ，大家可以搵數據上一上手，處理好 pre processing 同大概理解下 regression 的結果 (我無show d stats，不過可能下次會)，下篇文我再講多一個連systematic quant and trader 都可以利用到的 alpha。
另外，進階plan有需要問我攞分鐘data去測試的，可以message我。Codes我放左係篇文最底。加油!
Cheers,
阿程
(Alpha - 日內交易(上篇))
import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
import seaborn as sns
from datetime import datetime, time
import mpld3
from mpld3 import plugins
def load_and_process_data(file_path, ticker):
"""
Load and process 1-minute stock data from txt file.
"""
# Read data from txt file
df = pd.read_csv(file_path, header=None,
names=['datetime', 'open', 'high', 'low', 'close', 'volume'])
# Convert datetime to proper format
df['datetime'] = pd.to_datetime(df['datetime'])
df.set_index('datetime', inplace=True)
# Filter for trading hours (9:30 to 16:00)
df = df.between_time('09:30', '16:00')
# Create a complete range of 1-minute timestamps
trading_days = np.unique(df.index.date)
all_minutes = []
for day in trading_days:
trading_start = pd.Timestamp.combine(day, time(9, 30))
trading_end = pd.Timestamp.combine(day, time(16, 0))
day_minutes = pd.date_range(start=trading_start, end=trading_end, freq='1min')
all_minutes.extend(day_minutes)
# Create new DataFrame with all minute intervals
full_df = pd.DataFrame(index=pd.DatetimeIndex(all_minutes))
# Join the actual data
full_df = full_df.join(df)
# First fill close with forward fill and backward fill
full_df['close'] = full_df['close'].fillna(method='ffill').fillna(method='bfill')
# Now use close to fill missing values in other OHLC columns
for col in ['open', 'high', 'low']:
# First try to forward fill values that exist
full_df[col] = full_df[col].fillna(method='ffill')
# Then fill any remaining NAs with the close price
full_df[col] = full_df[col].fillna(full_df['close'])
# Fill NA with 0 for volume
full_df['volume'] = full_df['volume'].fillna(0)
# Add ticker column
full_df['ticker'] = ticker
return full_df
def resample_and_calculate_returns(df):
"""
Resample 1-minute data to 30-minute intervals and calculate returns.
"""
# Resample to 30-minute data
resampled = df.resample('30min').agg({
'open': 'first',
'high': 'max',
'low': 'min',
'close': 'last',
'volume': 'sum',
'ticker': 'first'
})
# Filter again to ensure we only have trading hours (9:30 to 16:00)
resampled = resampled.between_time('09:30', '16:00')
# Calculate returns
resampled['returns'] = resampled['close'].pct_change()
return resampled
def analyze_intraday_patterns(df, n_days=50):
"""
Analyze patterns between 30-minute returns at time t and t+1.
Include closing prices and timestamps for interactive display.
Exclude 15:30:00 time interval.
"""
# Extract trading days
trading_days = np.unique(df.index.date)
# Get the most recent N days
recent_days = trading_days[-n_days:] if len(trading_days) > n_days else trading_days
# Filter data for recent days
date_mask = np.isin(df.index.date, recent_days)
recent_data = df[date_mask].copy()
# Group by day and time
results = []
for day in recent_days:
# Create a mask for this specific day
day_mask = np.array([d == day for d in recent_data.index.date])
day_data = recent_data[day_mask].sort_index()
if len(day_data) < 2: # Need at least 2 intervals for t and t+1
continue
# For each time interval except the last one of the day
for i in range(len(day_data) - 1):
# Skip if current time is 15:30:00
if day_data.index[i].time() == time(15, 30):
continue
t_row = day_data.iloc[i]
t_plus_1_row = day_data.iloc[i + 1]
# Skip if next interval is on a different day (end of trading day)
if day_data.index[i].date() != day_data.index[i+1].date():
continue
if np.isnan(t_row['returns']) or np.isnan(t_plus_1_row['returns']):
continue
results.append({
'day': day,
'timestamp_t': day_data.index[i],
'timestamp_t_plus_1': day_data.index[i + 1],
'interval_t': day_data.index[i].time(),
'interval_t_plus_1': day_data.index[i + 1].time(),
'close_t': t_row['close'],
'close_t_plus_1': t_plus_1_row['close'],
'return_t': t_row['returns'],
'return_t_plus_1': t_plus_1_row['returns']
})
return pd.DataFrame(results)
def print_interval_data(patterns_df):
"""
Print detailed data for each time interval.
Exclude 15:30:00 time interval.
"""
if patterns_df.empty:
print("No patterns data available")
return
# Group by interval_t
grouped = patterns_df.groupby('interval_t')
# Filter intervals to include only those within trading hours (9:30 to 15:30)
# Exclude 15:30:00!!! This is very important otherwise you got a fake useless returns!
trading_start = time(9, 30)
trading_end = time(15, 30) # Changed from 16:00 to 15:30
intervals = sorted([
interval for interval in grouped.groups.keys()
if trading_start <= interval < trading_end # Changed <= to < to exclude 15:30
])
# For each interval
for interval in intervals:
interval_data = grouped.get_group(interval)
# Check that next interval is also within trading hours
next_interval = interval_data['interval_t_plus_1'].iloc[0] if not interval_data.empty else None
if next_interval is None or next_interval > trading_end:
continue
# Calculate correlation
corr = interval_data['return_t'].corr(interval_data['return_t_plus_1'])
print(f"\n{'='*80}")
print(f"Time Interval: {interval} → {next_interval} (Correlation: {corr:.3f})")
print(f"{'='*80}")
print(f"{'Day':<12} {'Timestamp t':<22} {'Close t':>10} {'Return t':>10} {'Timestamp t+1':<22} {'Close t+1':>10} {'Return t+1':>10}")
print(f"{'-'12} {'-'22} {'-'10} {'-'10} {'-'22} {'-'10} {'-'*10}")
# Sort by date for better readability
sorted_data = interval_data.sort_values('day')
for , row in sorteddata.iterrows():
print(f"{row['day'].strftime('%Y-%m-%d'):<12} "
f"{row['timestamp_t'].strftime('%Y-%m-%d %H:%M:%S'):<22} "
f"${row['close_t']:>9.2f} "
f"{row['return_t']:>9.4f} "
f"{row['timestamp_t_plus_1'].strftime('%Y-%m-%d %H:%M:%S'):<22} "
f"${row['close_t_plus_1']:>9.2f} "
f"{row['return_t_plus_1']:>9.4f}")
# Print summary statistics
print(f"\nSummary Statistics:")
print(f"Avg Close t: ${interval_data['close_t'].mean():.2f}")
print(f"Avg Close t+1: ${interval_data['close_t_plus_1'].mean():.2f}")
print(f"Avg Return t: {interval_data['return_t'].mean():.4f}")
print(f"Avg Return t+1: {interval_data['return_t_plus_1'].mean():.4f}")
print(f"Std Return t: {interval_data['return_t'].std():.4f}")
print(f"Std Return t+1: {interval_data['return_t_plus_1'].std():.4f}")
def plot_return_correlations_interactive(patterns_df, ticker):
"""
Plot interactive correlations between returns at time t and t+1.
Exclude 15:30:00 time interval.
"""
if patterns_df.empty:
print(f"No patterns data for {ticker}")
return None
# Group by interval_t
grouped = patterns_df.groupby('interval_t')
# Filter intervals to include only those within trading hours (9:30 to 15:30)
# Exclude 15:30:00
trading_start = time(9, 30)
trading_end = time(15, 30) # Changed from 16:00 to 15:30
intervals = sorted([
interval for interval in grouped.groups.keys()
if trading_start <= interval < trading_end # Changed <= to < to exclude 15:30
])
# Calculate number of plots
n_plots = len(intervals)
print(f"Number of plots for {ticker}: {n_plots}")
if n_plots == 0:
print(f"No intervals within trading hours for {ticker}")
return None
# Create a grid of scatter plots
fig_rows = int(np.ceil(n_plots / 3))
fig, axes = plt.subplots(fig_rows, 3, figsize=(18, 4 * fig_rows))
# Handle different dimensions of axes
if fig_rows == 1 and n_plots <= 3:
if n_plots == 1:
axes = np.array([axes]) # Force to array for indexing
else:
axes = axes.flatten()
for i, interval in enumerate(intervals):
# Create mask for this interval
interval_mask = np.array([t == interval for t in patterns_df['interval_t']])
interval_data = patterns_df[interval_mask]
# Check that next interval is also within trading hours
next_interval = interval_data['interval_t_plus_1'].iloc[0] if not interval_data.empty else None
if next_interval is None or next_interval > trading_end:
continue
# Get the axis for this plot
ax = axes[i % len(axes)]
# Calculate correlation
corr = interval_data['return_t'].corr(interval_data['return_t_plus_1'])
# Create scatter plot with regression line
sns.regplot(x='return_t', y='return_t_plus_1', data=interval_data, ax=ax, scatter_kws={'alpha':0.5})
# Add interactive hover points
scatter = ax.collections[0]
# Create tooltip labels
labels = []
for , row in intervaldata.iterrows():
tooltip = (
f"Day: {row['day'].strftime('%Y-%m-%d')}<br>"
f"Time {row['interval_t']}: Close ${row['close_t']:.2f}, Return: {row['return_t']:.4f}<br>"
f"Time {row['interval_t_plus_1']}: Close ${row['close_t_plus_1']:.2f}, Return: {row['return_t_plus_1']:.4f}<br>"
f"Timestamps: {row['timestamp_t']} → {row['timestamp_t_plus_1']}"
)
labels.append(tooltip)
tooltip = plugins.PointHTMLTooltip(scatter, labels, voffset=10, hoffset=10)
plugins.connect(fig, tooltip)
ax.set_title(f'{interval} → {next_interval}: Correlation = {corr:.3f}')
ax.set_xlabel(f'Return at {interval}')
ax.set_ylabel(f'Return at {next_interval}')
# Hide any unused subplots
for j in range(n_plots, len(axes)):
axes[j].set_visible(False)
fig.suptitle(f'{ticker}: Correlation between t and t+1 30-minute returns (Excluding 15:30-16:00)')
plt.tight_layout()
fig.subplots_adjust(top=0.95)
return fig
def main():
"""
Main function to run the analysis.
"""
try:
# Define file path to LLY txt file
lly_file = 'LLY_1min.txt' # Update this with your actual file path
print("Processing LLY.US data from file...")
lly_df = load_and_process_data(lly_file, 'LLY.US')
# Calculate returns
print("Calculating 30-minute returns...")
lly_returns = resample_and_calculate_returns(lly_df)
# Display sample of resampled data
print("\nLLY.US 30-minute data sample:")
print(lly_returns.head())
# Analyze patterns
print("\nAnalyzing intraday patterns (excluding 15:30-16:00 interval)...")
lly_patterns = analyze_intraday_patterns(lly_returns, n_days=200)
# Print detailed data for each interval
print("\nPrinting detailed data for each time interval:")
print_interval_data(lly_patterns)
# Plot correlations
print("\nPlotting correlations...")
lly_fig = plot_return_correlations_interactive(lly_patterns, 'LLY.US')
# Generate HTML for interactive plot
if lly_fig:
html_out = mpld3.fig_to_html(lly_fig)
with open('LLY_interactive_correlations.html', 'w') as f:
f.write(html_out)
print("Interactive HTML saved to LLY_interactive_correlations.html")
# Also save static version
lly_fig.savefig('LLY_return_correlations.png', dpi=300, bbox_inches='tight')
print("Static figure saved to LLY_return_correlations.png")
plt.show()
except Exception as e:
print(f"An error occurred: {e}")
import traceback
traceback.print_exc() # Print the full stack trace for better debugging
if __name__ == "__main__":
main()
Comments
想問個resample data 既問題 就好似resample price 應該用average 定係 extract last row value 咁 如果對住啲non price data 例如 market sentiment/ supply change 呢啲 係咪case by case
Samuel
2025-03-16 02:58:22 +0000 UTC1. 而個係我自己個人preference，我中意granular d build up，因為有時d 低頻數據會detect 錯 high / low， 但現實根本做唔到。 2. 支持阻力係execution上問題，通常去到而d位 spread + slippage 會變大，除非infra好，低latency做到 exact pricing，唔係日績月累，個交易成本會好高，搞到回測反映唔到實況。 3. You can try，都係 depends on infra
SY
2025-03-07 12:13:14 +0000 UTCthanks for sharing. 有幾個問題想問： 1. 點解用1m 再resample到30m 而唔直接用30m interval 嘅？ 有咩分別？ 2. 點解通常用時間比用支持阻力位好？背後有咩邏輯？ 3. 如果1m 即使捕捉到個regression但因為expected returns 太低，改成1h / 1d 係咪會好啲？（provided a higher correlation)
dennis lee
2025-03-07 09:43:06 +0000 UTC
