常見問題

import finlab
from finlab import data

# 1. 確認已登入
finlab.login()

# 2. 搜尋正確的欄位名稱
print(data.search('收盤'))

# 3. 用正確名稱重新下載
close = data.get('price:收盤價')
print(f"資料範圍：{close.index[0]} ~ {close.index[-1]}")

from finlab import data

close = data.get('price:收盤價')

if close.empty:
    print("資料為空：檢查 universe 篩選條件是否過嚴，或資料表本身是否沒有資料")
else:
    print(f"資料正常：{close.shape[0]} 個交易日，{close.shape[1]} 檔股票")

import pandas as pd
from finlab import data
from finlab.dataframe import FinlabDataFrame

basic_info = data.get('company_basic_info').set_index('stock_id')
營業收入 = data.get('financial_statement:營業收入淨額')
流通在外股數 = basic_info['已發行普通股數或TDR原發行股數']
流通在外股數 = FinlabDataFrame(
    pd.DataFrame([list(流通在外股數)], columns=basic_info.index, index=營業收入.index)
)

from finlab import data

close = data.get('price:收盤價')  # 日資料
roa = data.get('fundamental_features:ROA稅後息前')  # 季資料

# 不需要這樣寫
roa_daily = roa.reindex(close.index, method='ffill')
condition = (close < 30) & (roa_daily > 0)

# 直接寫就好
condition = (close < 30) & (roa > 0)

from finlab import data

close = data.get('price:收盤價')
asset = data.get('financial_statement:資產總額')

# FinlabDataFrame 會在運算時自動對齊不同頻率的資料
asset_growth = asset / asset.shift(4) - 1

# 要把最後的 position 換成交易日頻率，再對 position 做 reindex
position = (asset_growth > 0).index_str_to_date()
position = position.reindex(close.index, method='ffill')

df = df.index_str_to_date()  # 依各股實際公布日
df = df.deadline()  # 依統一公告截止日

import pandas as pd
from finlab.dataframe import FinlabDataFrame

raw = pd.read_csv('factor.csv', index_col=0)
raw.index = pd.to_datetime(raw.index)

factor = FinlabDataFrame(raw)

month_condition = close.index.to_series().apply(lambda x: x.month not in [7, 8])

report = sim(position, stop_loss=0.1)

from finlab.backtest import sim

# 檢查進場訊號
entry_stats = position.sum(axis=1)
print(f"平均每日進場股票數：{entry_stats.mean():.2f}")
print(f"最大進場股票數：{entry_stats.max()}")

if entry_stats.mean() < 1:
    # 進場條件過嚴：放寬條件（如價格區間），或用 .is_largest(N) 固定檔數
    # 用分數排名、不是用條件：每天都選出相對 20 日均線最強的 30 檔
    ma20 = close.average(20)
    position_fixed = (close / ma20).is_largest(30)

report = sim(position, resample='M')
trades = report.get_trades()
print(f"{len(trades)} 筆交易" if len(trades) else "沒有交易紀錄，請放寬進場條件")

report = sim(position, upload=False)

import os
from finlab.online.sinopac_account import SinopacAccount

# 檢查環境變數
required_vars = [
    'SHIOAJI_API_KEY',
    'SHIOAJI_SECRET_KEY',
    'SHIOAJI_CERT_PATH',
    'SHIOAJI_CERT_PASSWORD',
]
for var in required_vars:
    if not os.environ.get(var):
        print(f"{var} 未設定")

# 檢查憑證檔案
cert_path = os.environ.get('SHIOAJI_CERT_PATH')
if cert_path and not os.path.exists(cert_path):
    print(f"憑證檔案不存在：{cert_path}")

# 嘗試連線
try:
    acc = SinopacAccount()
    print("帳戶連線成功")
except Exception as e:
    print(f"連線失敗：{e}，請檢查 API key、憑證路徑與密碼")

from finlab.online.order_executor import OrderExecutor, Position

strategy = Position.from_report(report, 1_000_000)  # 這個策略的部位，投入 100 萬
manual = Position({'2330': 1})  # 手單部位：自己另外持有的一張 2330

executor = OrderExecutor(strategy + manual, account=account)
executor.create_orders(view_only=True)  # 先預覽

from finlab.online.order_executor import OrderExecutor, Position

fund = 1_000_000  # 要配置給這個策略的資金（新台幣）

available_cash = account.get_cash()
print(f"可用資金：NT$ {available_cash:,.0f}")
print(f"預計配置：NT$ {fund:,.0f}")

if fund > available_cash * 0.9:  # 留 10% 緩衝
    print("資金不足：降低配置資金、追加資金，或減少持股數量")
else:
    position = Position.from_report(report, fund)
    executor = OrderExecutor(position, account=account)
    executor.create_orders(view_only=True)  # 先預覽

from finlab.online.order_executor import OrderExecutor

executor = OrderExecutor(position, account=account)

# 1. 列出處置股與警示股
executor.show_alerting_stocks()

# 2. 先預覽
executor.create_orders(view_only=True)

# 3. 確認後執行
if input("確認無誤後輸入 'YES' 執行下單：") == 'YES':
    executor.create_orders()

import os
import pandas as pd
from finlab.market import Market

class CryptoMarket(Market):
    def __init__(self, data_dir='./data'):
        self.data_dir = data_dir

    def get_price(self, trade_at_price='close', adj=True):
        file_path = os.path.join(self.data_dir, f'crypto_{trade_at_price}.csv')

        if not os.path.exists(file_path):
            raise FileNotFoundError(
                f"找不到資料檔案：{file_path}\n"
                f"請確認檔案在 {self.data_dir} 目錄、檔名正確，"
                f"目前工作目錄：{os.getcwd()}"
            )

        return pd.read_csv(file_path, index_col=0, parse_dates=True)

try:
    market = CryptoMarket(data_dir='/path/to/data')
    close = market.get_price('close')
except FileNotFoundError as e:
    print(e)

import pandas as pd

df = pd.read_csv('data.csv', index_col=0)

# 把 index 轉成 DatetimeIndex
if not isinstance(df.index, pd.DatetimeIndex):
    try:
        df.index = pd.to_datetime(df.index)
    except Exception as e:
        print(f"無法轉換 index：{e}")
        print(f"index 範例：{df.index[:3].tolist()}")
        print("請確認第一欄是日期格式（如 2020-01-01）")

# 把非數值欄位轉成數值
non_numeric_cols = df.select_dtypes(exclude=['number']).columns.tolist()
for col in non_numeric_cols:
    df[col] = pd.to_numeric(df[col], errors='coerce')

from finlab import data
from finlab.ml import feature as mlf, label as mll

close = data.get('price:收盤價')
features = mlf.combine({'momentum': close / close.shift(20)}, resample='W')
labels = mll.excess_over_mean(features.index, resample='W')

# 特徵與標籤對齊
data_ml = features.join(labels.rename('label'), how='inner')

# 切分訓練集
train_data = data_ml[data_ml.index.get_level_values(0) <= '2022-12-31']
print(f"訓練集：{len(train_data)} 筆")

train_end = '2022-12-31'
test_start = '2023-01-01'

train_data = data_ml[data_ml.index.get_level_values(0) <= train_end]
test_data = data_ml[data_ml.index.get_level_values(0) >= test_start]

train_last_date = train_data.index.get_level_values(0).max()
test_first_date = test_data.index.get_level_values(0).min()

if train_last_date >= test_first_date:
    raise ValueError(
        f"訓練集與測試集日期重疊：訓練集最後 {train_last_date}，"
        f"測試集最早 {test_first_date}，會造成資料洩露"
    )

import finlab.ml.qlib as q

train = train_data.dropna()
model = q.LGBModel()
model.fit(train.drop(columns=['label']), train['label'])

all_pred = model.predict(data_ml.drop(columns=['label']))

if all_pred.isna().all().all():
    X_pred = data_ml.drop(columns=['label'])
    print(f"NaN 數量：{X_pred.isna().sum().sum()}")
    print(f"Inf 數量：{(X_pred == float('inf')).sum().sum()}")

    # 補缺失值後重新預測
    X_pred = X_pred.fillna(0).replace([float('inf'), float('-inf')], 0)
    all_pred = model.predict(X_pred)

print(f"預測完成：{all_pred.notna().sum().sum()} 筆")

from google.colab import drive

drive.mount('/content/drive')