feat(openinsider): 新增 OpenInsider 內部人交易爬蟲,支援多標的與每日排程

- 新增 app/crawlers/openinsider.py,來源 http://openinsider.com/search?q={symbol}

- 支援多標的:以 SYMBOLS=PLTR,NVDA,... 同時追多檔(或使用 SYMBOL 單一)

- runner: 多實例排程與啟動;/check 會依序觸發全部爬蟲

- API: /info、/stats、/check、/notify_test 支援多爬蟲回應

- config/base: 新增 RUN_DAILY_AT 每日固定時間;未設定則用 CHECK_INTERVAL

- notifications: 新增 send_custom_email、send_text_webhook、send_text_discord

- README 與 .env.template 更新;.env 改為 CRAWLER_TYPE=openinsider

- 移除 quiver_insiders 爬蟲與相關設定

BREAKING CHANGE: 不再支援 CRAWLER_TYPE=quiver_insiders;請改用 openinsider。
This commit is contained in:
2025-09-04 22:32:29 +08:00
parent 58cc979b5b
commit e89567643b
8 changed files with 368 additions and 40 deletions

View File

@@ -42,6 +42,27 @@ def send_email(new_picks: List[Dict], cfg: EmailConfig) -> None:
server.quit()
def send_custom_email(subject: str, body: str, cfg: EmailConfig) -> None:
msg = MIMEMultipart()
msg['From'] = cfg.from_email
msg['To'] = cfg.to_email
msg['Subject'] = subject
msg.attach(MIMEText(body, 'plain', 'utf-8'))
if cfg.smtp_security == 'ssl':
server = smtplib.SMTP_SSL(cfg.smtp_server, cfg.smtp_port)
else:
server = smtplib.SMTP(cfg.smtp_server, cfg.smtp_port)
server.ehlo()
if cfg.smtp_security == 'starttls':
server.starttls()
server.ehlo()
server.login(cfg.username, cfg.password)
server.send_message(msg)
server.quit()
def send_webhook(new_picks: List[Dict], url: str) -> None:
message = f"🚨 發現 {len(new_picks)} 條新的 Barron's 股票推薦!\n\n"
for pick in new_picks[:5]:
@@ -53,6 +74,11 @@ def send_webhook(new_picks: List[Dict], url: str) -> None:
requests.post(url, json=payload, timeout=10)
def send_text_webhook(message: str, url: str) -> None:
payload = {"text": message}
requests.post(url, json=payload, timeout=10)
def send_discord(new_picks: List[Dict], webhook: str) -> None:
embed = {
"title": "📈 Barron's 新股票推薦",
@@ -69,6 +95,22 @@ def send_discord(new_picks: List[Dict], webhook: str) -> None:
requests.post(webhook, json={"embeds": [embed]}, timeout=10)
def send_text_discord(title: str, description: str, lines: List[str], webhook: str) -> None:
embed = {
"title": title,
"description": description,
"color": 0x00ff00,
"fields": [],
}
for line in lines[:10]:
embed["fields"].append({
"name": line[:256],
"value": "\u200b",
"inline": False,
})
requests.post(webhook, json={"embeds": [embed]}, timeout=10)
def build_test_pick() -> Dict:
return {
'title': f"[測試] Barron's 通知發送 - {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}",
@@ -76,4 +118,3 @@ def build_test_pick() -> Dict:
'scraped_at': datetime.now().isoformat(),
'hash': hashlib.md5(str(datetime.now().timestamp()).encode()).hexdigest()[:8],
}