ValueInvesting / app.py
william1324's picture
Update app.py
8fb682a verified
Raw
History Blame Contribute Delete
20 kB
import gradio as gr
import yfinance as yf
import pandas as pd
import numpy as np
import requests
from dataclasses import dataclass
from datetime import datetime
@dataclass(frozen=True)
class AVUVTWConfig:
min_market_cap: float = 3_000_000_000 # 30億
max_market_cap: float = 50_000_000_000 # 500億
min_adv20: float = 100_000_000 # 1億
min_price: float = 10
history_period: str = "24mo"
min_history_days: int = 300
lookback_days: int = 252
skip_days: int = 21
vol_window: int = 60
max_volatility: float = 0.90
min_roe: float = 0.05
max_debt_ratio: float = 0.60
CONFIG = AVUVTWConfig()
CUSTOM_CSS = """
body {
background: linear-gradient(135deg, #fbf7ec, #f4ead6, #efe2c4);
}
.gradio-container {
max-width: 100% !important;
margin: auto !important;
font-family: 'Noto Sans TC', sans-serif;
color: #111111;
padding: 10px !important;
}
.hero {
background: rgba(255,255,255,0.96);
border: 1px solid rgba(212,175,55,0.32);
border-radius: 22px;
padding: 22px;
margin-bottom: 16px;
box-shadow: 0 12px 28px rgba(120,90,30,0.10);
}
.title {
font-size: 32px;
font-weight: 950;
color: #111111;
line-height: 1.2;
}
.subtitle {
margin-top: 10px;
font-size: 15px;
color: #555555;
line-height: 1.6;
}
.control-card {
background: rgba(255,255,255,0.95);
border-radius: 20px;
padding: 18px;
border: 1px solid rgba(212,175,55,0.25);
box-shadow: 0 10px 24px rgba(120,90,30,0.08);
}
.result-card {
background: rgba(255,255,255,0.94);
border-radius: 20px;
padding: 14px;
border: 1px solid rgba(212,175,55,0.20);
}
button {
min-height: 52px !important;
font-size: 17px !important;
font-weight: 800 !important;
border-radius: 16px !important;
}
label, .wrap {
font-size: 15px !important;
}
textarea, input {
font-size: 16px !important;
}
/* 表格手機橫向滑動 */
.dataframe {
overflow-x: auto !important;
}
/* 手機版 */
@media screen and (max-width: 768px) {
.gradio-container {
padding: 8px !important;
}
.hero {
padding: 18px;
border-radius: 18px;
}
.title {
font-size: 26px;
text-align: left;
}
.subtitle {
font-size: 14px;
text-align: left;
}
.control-card {
padding: 14px;
border-radius: 18px;
}
.result-card {
padding: 10px;
border-radius: 16px;
}
button {
width: 100% !important;
min-height: 56px !important;
font-size: 18px !important;
}
.tab-nav button {
font-size: 15px !important;
}
table {
font-size: 13px !important;
white-space: nowrap !important;
}
}
"""
def fetch_twse_symbols():
url = "https://openapi.twse.com.tw/v1/opendata/t187ap03_L"
try:
data = requests.get(url, timeout=12).json()
rows = []
for item in data:
code = str(item.get("公司代號", "")).strip()
name = str(item.get("公司名稱", "")).strip()
if code.isdigit():
rows.append({
"symbol": f"{code}.TW",
"code": code,
"name": name,
"market": "上市"
})
return pd.DataFrame(rows)
except Exception:
return pd.DataFrame()
def fetch_tpex_symbols():
url = "https://www.tpex.org.tw/openapi/v1/mopsfin_t187ap03_O"
try:
data = requests.get(url, timeout=12).json()
rows = []
for item in data:
code = str(item.get("SecuritiesCompanyCode", "")).strip()
name = str(item.get("CompanyName", "")).strip()
if code.isdigit():
rows.append({
"symbol": f"{code}.TWO",
"code": code,
"name": name,
"market": "上櫃"
})
return pd.DataFrame(rows)
except Exception:
return pd.DataFrame()
def get_universe(market_choice):
twse = fetch_twse_symbols()
tpex = fetch_tpex_symbols()
if market_choice == "上市":
df = twse
elif market_choice == "上櫃":
df = tpex
else:
df = pd.concat([twse, tpex], ignore_index=True)
if df.empty:
fallback = [
("2458.TW", "義隆", "上市"),
("3034.TW", "聯詠", "上市"),
("3189.TW", "景碩", "上市"),
("3533.TW", "嘉澤", "上市"),
("3706.TW", "神達", "上市"),
("4919.TW", "新唐", "上市"),
("6278.TW", "台表科", "上市"),
("3105.TWO", "穩懋", "上櫃"),
("3264.TWO", "欣銓", "上櫃"),
("6187.TWO", "萬潤", "上櫃"),
("6223.TWO", "旺矽", "上櫃"),
("6488.TWO", "環球晶", "上櫃"),
]
df = pd.DataFrame(fallback, columns=["symbol", "name", "market"])
df["code"] = df["symbol"].str[:4]
return df.drop_duplicates("symbol").reset_index(drop=True)
def safe_float(x, default=np.nan):
try:
if x is None:
return default
return float(x)
except Exception:
return default
def get_basic_info(symbol):
try:
ticker = yf.Ticker(symbol)
info = ticker.info or {}
fast = {}
try:
fast = ticker.fast_info or {}
except Exception:
pass
market_cap = safe_float(fast.get("market_cap", np.nan))
if pd.isna(market_cap):
market_cap = safe_float(info.get("marketCap", np.nan))
return {
"market_cap": market_cap,
"pe": safe_float(info.get("trailingPE", np.nan)),
"pb": safe_float(info.get("priceToBook", np.nan)),
"roe": safe_float(info.get("returnOnEquity", np.nan)),
"roa": safe_float(info.get("returnOnAssets", np.nan)),
"gross_margin": safe_float(info.get("grossMargins", np.nan)),
"operating_margin": safe_float(info.get("operatingMargins", np.nan)),
"debt_to_equity": safe_float(info.get("debtToEquity", np.nan)),
"free_cashflow": safe_float(info.get("freeCashflow", np.nan)),
}
except Exception:
return {
"market_cap": np.nan,
"pe": np.nan,
"pb": np.nan,
"roe": np.nan,
"roa": np.nan,
"gross_margin": np.nan,
"operating_margin": np.nan,
"debt_to_equity": np.nan,
"free_cashflow": np.nan,
}
def get_price_metrics(symbol, config):
try:
df = yf.download(
symbol,
period=config.history_period,
interval="1d",
auto_adjust=True,
progress=False,
threads=False,
)
if df is None or df.empty:
return None
if isinstance(df.columns, pd.MultiIndex):
df.columns = df.columns.get_level_values(0)
if len(df) < config.min_history_days:
return None
close = df["Close"].dropna()
volume = df["Volume"].dropna()
if len(close) < config.min_history_days:
return None
price = float(close.iloc[-1])
if price < config.min_price:
return None
amount = close * volume
adv20 = float(amount.rolling(20).mean().iloc[-1])
if pd.isna(adv20) or adv20 < config.min_adv20:
return None
recent = close.iloc[-config.skip_days]
past = close.iloc[-config.lookback_days]
mom_12_1 = float(recent / past - 1)
ret_6m = float(close.iloc[-1] / close.iloc[-126] - 1) if len(close) > 126 else np.nan
ret_3m = float(close.iloc[-1] / close.iloc[-63] - 1) if len(close) > 63 else np.nan
daily_ret = close.pct_change().dropna()
volatility = float(daily_ret.rolling(config.vol_window).std().iloc[-1] * np.sqrt(252))
ma200 = float(close.rolling(200).mean().iloc[-1])
above_ma200 = bool(price > ma200)
diff = close.diff()
up_days = float((diff > 0).rolling(config.lookback_days).sum().iloc[-1])
down_days = float((diff < 0).rolling(config.lookback_days).sum().iloc[-1])
if up_days + down_days > 0:
fip = (up_days - down_days) / (up_days + down_days)
else:
fip = np.nan
rolling_max = close.rolling(252).max()
drawdown = close / rolling_max - 1
max_drawdown = float(drawdown.rolling(252).min().iloc[-1])
return {
"price": price,
"adv20": adv20,
"mom_12_1": mom_12_1,
"ret_6m": ret_6m,
"ret_3m": ret_3m,
"volatility": volatility,
"above_ma200": above_ma200,
"fip": fip,
"max_drawdown": max_drawdown,
}
except Exception:
return None
def analyze_one_stock(row, config):
symbol = row["symbol"]
price_metrics = get_price_metrics(symbol, config)
if price_metrics is None:
return None
basic = get_basic_info(symbol)
market_cap = basic["market_cap"]
exclude_reasons = []
if pd.isna(market_cap):
exclude_reasons.append("無市值資料")
else:
if market_cap < config.min_market_cap:
exclude_reasons.append("市值低於30億")
if market_cap > config.max_market_cap:
exclude_reasons.append("市值高於500億")
if price_metrics["adv20"] < config.min_adv20:
exclude_reasons.append("20日均成交值低於1億")
if price_metrics["volatility"] > config.max_volatility:
exclude_reasons.append("波動率過高")
roe = basic["roe"]
if not pd.isna(roe) and roe < config.min_roe:
exclude_reasons.append("ROE低於5%")
operating_margin = basic["operating_margin"]
if not pd.isna(operating_margin) and operating_margin < 0:
exclude_reasons.append("營業利益率為負")
debt_to_equity = basic["debt_to_equity"]
if not pd.isna(debt_to_equity):
debt_ratio_like = debt_to_equity / 100
if debt_ratio_like > config.max_debt_ratio:
exclude_reasons.append("負債偏高")
free_cashflow = basic["free_cashflow"]
if not pd.isna(free_cashflow) and free_cashflow < 0:
exclude_reasons.append("自由現金流為負")
if price_metrics["mom_12_1"] < -0.20:
exclude_reasons.append("12-1月動能過弱")
if not price_metrics["above_ma200"]:
exclude_reasons.append("跌破MA200")
return {
"symbol": row["symbol"],
"code": row["code"],
"name": row["name"],
"market": row["market"],
**price_metrics,
**basic,
"exclude_reason": "、".join(exclude_reasons),
}
def percentile_score(series, higher_is_better=True):
s = pd.to_numeric(series, errors="coerce")
if higher_is_better:
return s.rank(pct=True)
return 1 - s.rank(pct=True)
def build_ranking(df, top_n):
df = df.copy()
valid = df[df["exclude_reason"] == ""].copy()
if valid.empty:
return pd.DataFrame()
for col in [
"pe", "pb", "roe", "roa", "gross_margin", "operating_margin",
"mom_12_1", "fip", "volatility", "market_cap", "adv20"
]:
valid[col] = pd.to_numeric(valid[col], errors="coerce")
valid["pe_score"] = percentile_score(valid["pe"], higher_is_better=False)
valid["pb_score"] = percentile_score(valid["pb"], higher_is_better=False)
valid["value_score"] = (
0.50 * valid["pe_score"].fillna(0.5)
+ 0.50 * valid["pb_score"].fillna(0.5)
)
valid["roe_score"] = percentile_score(valid["roe"], higher_is_better=True)
valid["op_margin_score"] = percentile_score(valid["operating_margin"], higher_is_better=True)
valid["gross_margin_score"] = percentile_score(valid["gross_margin"], higher_is_better=True)
valid["roa_score"] = percentile_score(valid["roa"], higher_is_better=True)
valid["quality_score"] = (
0.35 * valid["roe_score"].fillna(0.5)
+ 0.30 * valid["op_margin_score"].fillna(0.5)
+ 0.20 * valid["gross_margin_score"].fillna(0.5)
+ 0.15 * valid["roa_score"].fillna(0.5)
)
valid["momentum_score"] = percentile_score(valid["mom_12_1"], higher_is_better=True).fillna(0.5)
valid["fip_score"] = percentile_score(valid["fip"], higher_is_better=True).fillna(0.5)
valid["low_vol_score"] = percentile_score(valid["volatility"], higher_is_better=False).fillna(0.5)
valid["avuv_tw_score"] = (
0.30 * valid["value_score"].fillna(0.5)
+ 0.30 * valid["quality_score"].fillna(0.5)
+ 0.20 * valid["momentum_score"]
+ 0.10 * valid["fip_score"]
+ 0.10 * valid["low_vol_score"]
)
valid = valid.sort_values("avuv_tw_score", ascending=False).reset_index(drop=True)
valid["rank"] = np.arange(1, len(valid) + 1)
ranking = valid.head(int(top_n)).copy()
ranking["market_cap_億"] = ranking["market_cap"] / 100_000_000
ranking["adv20_億"] = ranking["adv20"] / 100_000_000
output = ranking[[
"rank",
"symbol",
"name",
"market",
"avuv_tw_score",
"value_score",
"quality_score",
"momentum_score",
"fip_score",
"low_vol_score",
"price",
"market_cap_億",
"adv20_億",
"pe",
"pb",
"roe",
"roa",
"gross_margin",
"operating_margin",
"mom_12_1",
"ret_6m",
"ret_3m",
"volatility",
"max_drawdown"
]]
output = output.rename(columns={
"rank": "排名",
"symbol": "股票代號",
"name": "股票名稱",
"market": "市場",
"avuv_tw_score": "AVUV-TW總分",
"value_score": "價值分數",
"quality_score": "品質分數",
"momentum_score": "動能分數",
"fip_score": "FIP分數",
"low_vol_score": "低波動分數",
"price": "股價",
"market_cap_億": "市值_億",
"adv20_億": "20日均成交值_億",
"pe": "本益比",
"pb": "股價淨值比",
"roe": "ROE",
"roa": "ROA",
"gross_margin": "毛利率",
"operating_margin": "營業利益率",
"mom_12_1": "12-1月動能",
"ret_6m": "6月報酬",
"ret_3m": "3月報酬",
"volatility": "年化波動率",
"max_drawdown": "最大回撤",
})
num_cols = output.select_dtypes(include=[np.number]).columns
output[num_cols] = output[num_cols].round(4)
return output
def run_strategy(market_choice, max_scan_count, top_n):
universe = get_universe(market_choice)
if universe.empty:
return pd.DataFrame(), pd.DataFrame(), "無法取得股票池。"
universe = universe.head(int(max_scan_count)).copy()
rows = []
for _, row in universe.iterrows():
result = analyze_one_stock(row, CONFIG)
if result is not None:
rows.append(result)
if not rows:
return pd.DataFrame(), pd.DataFrame(), "沒有股票通過基本價格與成交量資料檢查。"
raw = pd.DataFrame(rows)
ranking = build_ranking(raw, top_n)
raw_display = raw.copy()
raw_display["market_cap_億"] = raw_display["market_cap"] / 100_000_000
raw_display["adv20_億"] = raw_display["adv20"] / 100_000_000
raw_display = raw_display[[
"symbol",
"name",
"market",
"price",
"market_cap_億",
"adv20_億",
"pe",
"pb",
"roe",
"roa",
"gross_margin",
"operating_margin",
"debt_to_equity",
"free_cashflow",
"mom_12_1",
"ret_6m",
"ret_3m",
"volatility",
"fip",
"above_ma200",
"max_drawdown",
"exclude_reason"
]]
raw_display = raw_display.rename(columns={
"symbol": "股票代號",
"name": "股票名稱",
"market": "市場",
"price": "股價",
"market_cap_億": "市值_億",
"adv20_億": "20日均成交值_億",
"pe": "本益比",
"pb": "股價淨值比",
"roe": "ROE",
"roa": "ROA",
"gross_margin": "毛利率",
"operating_margin": "營業利益率",
"debt_to_equity": "負債權益比",
"free_cashflow": "自由現金流",
"mom_12_1": "12-1月動能",
"ret_6m": "6月報酬",
"ret_3m": "3月報酬",
"volatility": "年化波動率",
"fip": "FIP",
"above_ma200": "是否站上MA200",
"max_drawdown": "最大回撤",
"exclude_reason": "排除原因"
})
num_cols = raw_display.select_dtypes(include=[np.number]).columns
raw_display[num_cols] = raw_display[num_cols].round(4)
passed_count = len(raw[raw["exclude_reason"] == ""])
msg = f"""
### 分析完成
完成時間:{datetime.now().strftime("%Y-%m-%d %H:%M:%S")}
掃描市場:{market_choice}
掃描股票數:{len(universe)}
成功取得資料:{len(raw)}
通過 AVUV-TW 條件:{passed_count}
篩選條件:
- 市值:30億~500億
- 20日均成交值:大於1億
- 股價:大於10元
- ROE:至少5%
- 年化波動率:低於90%
- 避免負自由現金流、負營業利益率、跌破MA200、動能過弱
總分公式:
AVUV-TW總分
= 30% 價值分數
+ 30% 品質分數
+ 20% 動能分數
+ 10% FIP分數
+ 10% 低波動分數
"""
if ranking.empty:
msg += "\n\n目前沒有股票完全通過條件,可以放寬市值、成交值或 ROE 條件。"
return ranking, raw_display, msg
with gr.Blocks(css=CUSTOM_CSS, title="AVUV-TW 台股小型價值股排行榜") as demo:
gr.HTML("""
<div class="hero">
<div class="title">AVUV-TW 台股排行榜</div>
<div class="subtitle">
篩選市值 30億~500億、20日均成交值大於1億的台股,
並依照價值、品質、動能、FIP、低波動因子製作排行榜。
</div>
</div>
""")
with gr.Column():
with gr.Group(elem_classes="control-card"):
market_choice = gr.Radio(
choices=["上市", "上櫃", "上市+上櫃"],
value="上市+上櫃",
label="選擇市場"
)
max_scan_count = gr.Slider(
minimum=20,
maximum=1000,
value=200,
step=20,
label="最多掃描股票數"
)
top_n = gr.Slider(
minimum=5,
maximum=100,
value=30,
step=5,
label="排行榜顯示前 N 名"
)
run_btn = gr.Button("開始製作排行榜", variant="primary")
gr.Markdown("""
**固定條件**
市值:30億~500億
20日均成交值:大於1億
股價:大於10元
ROE:至少5%
年化波動率:小於90%
""")
with gr.Group(elem_classes="result-card"):
message_output = gr.Markdown()
with gr.Tabs():
with gr.Tab("排行榜"):
ranking_output = gr.Dataframe(
label="AVUV-TW 小型價值股排行榜",
interactive=False,
wrap=False
)
with gr.Tab("原始資料"):
raw_output = gr.Dataframe(
label="原始分析資料與排除原因",
interactive=False,
wrap=False
)
run_btn.click(
fn=run_strategy,
inputs=[market_choice, max_scan_count, top_n],
outputs=[ranking_output, raw_output, message_output]
)
if __name__ == "__main__":
demo.launch()