透過永豐證券 API 找出短線交易潛力股

import shioaji as sj

api = sj.Shioaji(simulation=True)  # 使用模擬環境即可

api.login(
    api_key="your_api_key", 
    secret_key="your_secret_key", 
    receive_window=60000
)

import datetime
import pandas as pd

# 設定查詢的起始日和結束日
start_date = (datetime.datetime.now() - datetime.timedelta(days=30)).strftime('%Y-%m-%d')
end_date = datetime.datetime.now().strftime('%Y-%m-%d')

kbars = api.kbars(
    contract=api.Contracts.Stocks['2330'],
    start=start_date,
    end=end_date,
)

df = pd.DataFrame({**kbars})
df.ts = pd.to_datetime(df.ts)
df['year'] = df['ts'].dt.year 
df['month'] = df['ts'].dt.month
df['day'] = df['ts'].dt.day

print(df.head(5))  # 顯示前五筆資料

import shioaji as sj
import history

# 登入永豐證券 API
api = sj.Shioaji(simulation=True)
api.login(
    api_key="your_api_key",
    secret_key="your_secret_key",
    receive_window=60000
)

# 執行篩選過程並將結果存為 Excel 檔案
hty = history.history_stock(api)
history_data = hty.return_df()
history_data.to_excel("stock.xlsx", index=False)

import pandas as pd
import datetime
import numpy as np

class history_stock:
    def __init__(self, api):
        self.api = api
        # 篩選股票價格範圍
        self.low_price = 100.5
        self.high_price = 139

        # 查詢開始日和結束日
        search_days = 30
        self.start_date = (datetime.datetime.now() - datetime.timedelta(days=search_days)).strftime('%Y-%m-%d')
        self.end_date = datetime.datetime.now().strftime('%Y-%m-%d')

        # 取得可當日沖銷的股票清單
        self.total_stock()
        # 計算各股開、高、收、低價及成交均量
        self.update_data()
        # 篩選符合條件的股票
        self.condition()

    def total_stock(self):
        item_name = []
        item_code = []
        for item in self.api.Contracts.Stocks.TSE:
            if item['day_trade'] == 'Yes':  # 篩選可當日沖銷的股票
                item_name.append(item['name'])
                item_code.append(item['code'])

        self.history_df = pd.DataFrame({'name': item_name, 'symbol': item_code})
        self.history_df['avg_volume'] = ''
        self.history_df['high'] = ''
        self.history_df['low'] = ''
        self.history_df['open'] = ''
        self.history_df['close'] = ''
        self.history_df['price_change'] = ''

    def update_data(self):
        for i in self.history_df.index:
            avg_volume, High, Low, Open, Close = self.kbar(self.history_df['symbol'].iloc[i], self.start_date, self.end_date)
            self.history_df['avg_volume'].iloc[i] = avg_volume
            self.history_df['high'].iloc[i] = High
            self.history_df['low'].iloc[i] = Low
            self.history_df['open'].iloc[i] = Open
            self.history_df['close'].iloc[i] = Close
            if Low is None:
                self.history_df['price_change'].iloc[i] = None
            else:
                self.history_df['price_change'].iloc[i] = (High - Low) / Low

        return self.history_df

    def kbar(self, symbol, start_date, end_date):
        kbars = self.api.kbars(
            contract=self.api.Contracts.Stocks[symbol],
            start=start_date,
            end=end_date,
        )

        df = pd.DataFrame({**kbars})
        df.ts = pd.to_datetime(df.ts)
        df['year'] = df['ts'].dt.year
        df['month'] = df['ts'].dt.month
        df['day'] = df['ts'].dt.day

        if len(df) == 0:
            return None, None, None, None, None
        else:
            High = df.groupby(['year', 'month', 'day']).max()['High'].iloc[-1]  # 最近一日最高價
            Low = df.groupby(['year', 'month', 'day']).min()['Low'].iloc[-1]    # 最近一日最低價
            df.sort_values(by=['year', 'month', 'day', 'ts'], ascending=[False, False, False, True], inplace=True)
            Open = df['Open'].iloc[0]
            df.sort_values(by=['year', 'month', 'day', 'ts'], ascending=[False, False, False, False], inplace=True)
            Close = df['Close'].iloc[0]
            df = df.groupby(['year', 'month', 'day'])[['Volume']].sum()
            return (sum(df['Volume']) / len(df)), High, Low, Open, Close

    def condition(self):
        # 篩選成交量大於 1000 和股價變動幅度大於 2% 的股票
        self.history_df = self.history_df.loc[np.logical_and(self.history_df.avg_volume >= 1000, self.history_df.price_change >= 0.02)]
        self.history_df = self.history_df[self.history_df.symbol.astype(str).str.len() == 4]  # 移除 ETF

        # 篩選股價在設定範圍內的股票
        price_series = self.history_df['close'].astype(float)
        price_range = np.logical_and(price_series >= self.low_price, price_series <= self.high_price)
        self.history_df = self.history_df[price_range]

        # 按照成交量排序
        self.history_df.sort_values(by="avg_volume", ascending=False, inplace=True)

    def return_df(self):
        return self.history_df