ddrwode/mini_code
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

dfefwef

Browse Source
This commit is contained in:
27942
2026-01-03 14:50:08 +08:00
parent 8c61b57d76
commit cecf176229
10 changed files with 9 additions and 231 deletions
Download Patch File Download Diff File Expand all files Collapse all files

144
polymarket 优化抓取.py
View File

@@ -1,144 +0,0 @@
import requests
import json
from concurrent.futures import ThreadPoolExecutor, as_completed
from typing import Any, Dict, List, Optional, Tuple
GAMMA = "https://gamma-api.polymarket.com"
CLOB = "https://clob.polymarket.com"
def parse_jsonish_list(v):
"""兼容 list 或 JSON 数组字符串。"""
if v is None:
return []
if isinstance(v, list):
return v
if isinstance(v, str):
s = v.strip()
if s.startswith("[") and s.endswith("]"):
return json.loads(s)
return [x.strip() for x in s.split(",") if x.strip()]
return []
def get_market_by_slug(session: requests.Session, slug: str) -> Dict[str, Any]:
r = session.get(f"{GAMMA}/markets/slug/{slug}", timeout=20)
r.raise_for_status()
return r.json()
def get_price(session: requests.Session, token_id: str, side: str) -> Optional[float]:
"""
side='buy' -> best bid
side='sell' -> best ask
"""
r = session.get(f"{CLOB}/price", params={"token_id": token_id, "side": side}, timeout=20)
r.raise_for_status()
p = r.json().get("price")
return float(p) if p is not None else None
def get_book(session: requests.Session, token_id: str) -> Dict[str, Any]:
"""包括 bids/asks 的订单簿快照。"""
r = session.get(f"{CLOB}/book", params={"token_id": token_id}, timeout=20)
r.raise_for_status()
return r.json()
def compute_mid_from_price(bid: Optional[float], ask: Optional[float]) -> Tuple[Optional[float], Optional[float]]:
if bid is None and ask is None:
return None, None
if bid is None:
return ask, None
if ask is None:
return bid, None
return (bid + ask) / 2.0, (ask - bid)
def trim_book(book: Dict[str, Any], top_n: int = 10) -> Dict[str, Any]:
bids = book.get("bids") or []
asks = book.get("asks") or []
# 原始返回一般是按价优排序；这里做一下保险排序
bids_sorted = sorted(bids, key=lambda x: float(x["price"]), reverse=True)[:top_n]
asks_sorted = sorted(asks, key=lambda x: float(x["price"]))[:top_n]
return {"bids": bids_sorted, "asks": asks_sorted}
def fetch_token_bundle(session: requests.Session, token_id: str, top_n: int = 10) -> Dict[str, Any]:
"""
对某个 token 并发抓取：
- best bid (/price?side=buy)
- best ask (/price?side=sell)
- order book (/book)
"""
with ThreadPoolExecutor(max_workers=3) as ex:
futures = {
ex.submit(get_price, session, token_id, "buy"): "bid",
ex.submit(get_price, session, token_id, "sell"): "ask",
ex.submit(get_book, session, token_id): "book",
}
out: Dict[str, Any] = {"token_id": token_id}
for fut in as_completed(futures):
key = futures[fut]
out[key] = fut.result()
bid = out.get("bid")
ask = out.get("ask")
mid, spread = compute_mid_from_price(bid, ask)
out["mid"] = mid
out["spread"] = spread
out["book"] = trim_book(out.get("book") or {}, top_n=top_n)
return out
def polymarket_updown(slug: str, decimals: int = 0, top_n: int = 10) -> Dict[str, Any]:
"""
返回：
- UP/DOWN 百分比（尽量贴网页）
- UP/DOWN 的 token_id、bid/ask/mid/spread
- UP/DOWN 订单簿 topN
"""
session = requests.Session()
session.headers.update({"User-Agent": "Mozilla/5.0"})
m = get_market_by_slug(session, slug)
outcomes = [str(x) for x in parse_jsonish_list(m.get("outcomes"))]
token_ids = [str(x) for x in parse_jsonish_list(m.get("clobTokenIds"))]
if len(token_ids) < 2:
raise RuntimeError(f"clobTokenIds missing/too short: {token_ids}")
token_map = dict(zip(outcomes, token_ids))
up_id = token_map.get("Up") or token_map.get("Yes") or token_ids[0]
dn_id = token_map.get("Down") or token_map.get("No") or token_ids[1]
# 并发抓 UP / DOWN 两个 token 的 bundle
with ThreadPoolExecutor(max_workers=2) as ex:
fut_up = ex.submit(fetch_token_bundle, session, up_id, top_n)
fut_dn = ex.submit(fetch_token_bundle, session, dn_id, top_n)
up = fut_up.result()
dn = fut_dn.result()
if up["mid"] is None or dn["mid"] is None:
return {
"error": "missing mid price (bid/ask both None?)",
"slug": slug,
"market_id": m.get("id"),
"question": m.get("question"),
"up": up,
"down": dn,
}
# 归一化到 100（防止 mid 不严格互补导致 up+down != 1）
s = float(up["mid"]) + float(dn["mid"])
up_pct = round(float(up["mid"]) / s * 100, decimals)
dn_pct = round(float(dn["mid"]) / s * 100, decimals)
return {
"question": m.get("question"),
"slug": m.get("slug"),
"market_id": m.get("id"),
"outcomes": outcomes,
"token_map_rough": token_map,
"up_pct": up_pct,
"down_pct": dn_pct,
"up": up,
"down": dn,
"note": "UP/DOWN% 使用 /price 的 bid/ask 计算 mid 并归一化；book 为订单簿 topN。",
}
if __name__ == "__main__":
print(polymarket_updown("eth-updown-15m-1767069900", decimals=0, top_n=10))

78
polymarket 抓取价格.py
View File

@@ -1,78 +0,0 @@
import requests
import json
GAMMA = "https://gamma-api.polymarket.com"
CLOB = "https://clob.polymarket.com"
def parse_jsonish_list(v):
if v is None:
return []
if isinstance(v, list):
return v
if isinstance(v, str):
return json.loads(v)
return []
def get_market_by_slug(slug: str):
r = requests.get(f"{GAMMA}/markets/slug/{slug}", timeout=20)
r.raise_for_status()
return r.json()
def get_best_price(token_id: str, side: str) -> float | None:
# side: "buy" => best bid, "sell" => best ask
r = requests.get(f"{CLOB}/price", params={"token_id": token_id, "side": side}, timeout=20)
r.raise_for_status()
p = r.json().get("price")
return float(p) if p is not None else None
def get_mid_or_fallback(token_id: str):
bid = get_best_price(token_id, "buy")
ask = get_best_price(token_id, "sell")
if bid is None and ask is None:
return None, bid, ask, None
# 如果只拿到一侧，就退化
if bid is None:
return ask, bid, ask, None
if ask is None:
return bid, bid, ask, None
spread = ask - bid
mid = (bid + ask) / 2
return mid, bid, ask, spread
def web_like_up_down(slug: str, decimals=0):
m = get_market_by_slug(slug)
outcomes = [str(x) for x in parse_jsonish_list(m.get("outcomes"))]
token_ids = [str(x) for x in parse_jsonish_list(m.get("clobTokenIds"))]
token_map = dict(zip(outcomes, token_ids))
up_id = token_map.get("Up") or token_map.get("Yes") or token_ids[0]
down_id = token_map.get("Down") or token_map.get("No") or token_ids[1]
up_mid, up_bid, up_ask, up_spread = get_mid_or_fallback(up_id)
dn_mid, dn_bid, dn_ask, dn_spread = get_mid_or_fallback(down_id)
if up_mid is None or dn_mid is None:
return {"error": "missing price", "up": (up_mid, up_bid, up_ask), "down": (dn_mid, dn_bid, dn_ask)}
# 归一化（避免因为点差导致 up+down != 1）
s = up_mid + dn_mid
up_pct = round(up_mid / s * 100, decimals)
dn_pct = round(dn_mid / s * 100, decimals)
return {
"question": m.get("question"),
"market_id": m.get("id"),
"slug": m.get("slug"),
"up_pct": up_pct,
"down_pct": dn_pct,
"debug": {
"up": {"token_id": up_id, "bid": up_bid, "ask": up_ask, "mid_used": up_mid, "spread": up_spread},
"down":{"token_id": down_id, "bid": dn_bid, "ask": dn_ask, "mid_used": dn_mid, "spread": dn_spread},
"note": "网页通常用 mid；若点差>0.10 则可能改用 last trade price。"
}
}
if __name__ == "__main__":
print(web_like_up_down("eth-updown-15m-1767069900", decimals=0))

BIN
polymarket.db
View File

Binary file not shown.

BIN
polymarket.db-shm
View File

Binary file not shown.

BIN
polymarket.db-wal
View File

Binary file not shown.

4
btc抓取.py → polymarket/btc抓取.py
View File

@@ -231,7 +231,7 @@ def save_snapshot(bucket_ts: int, fetched_at: int, raw: Dict[str, Any], db_path:
if __name__ == "__main__":
init_db("polymarket.db")
init_db("../polymarket.db")
while True:
data = fetch_current_market(
@@ -242,5 +242,5 @@ if __name__ == "__main__":
probe_offsets=(0, -900, 900), # 先试当前，再试前一档/后一档
)
save_snapshot(data["bucket_ts"], data["fetched_at"], data, "polymarket.db")
save_snapshot(data["bucket_ts"], data["fetched_at"], data, "../polymarket.db")
print("saved:", data["bucket_ts"], data["fetched_at"])

4
test1.py → polymarket/eth抓取.py
View File

@@ -231,7 +231,7 @@ def save_snapshot(bucket_ts: int, fetched_at: int, raw: Dict[str, Any], db_path:
if __name__ == "__main__":
init_db("polymarket.db")
init_db("../polymarket.db")
while True:
data = fetch_current_market(
@@ -242,5 +242,5 @@ if __name__ == "__main__":
probe_offsets=(0, -900, 900), # 先试当前，再试前一档/后一档
)
save_snapshot(data["bucket_ts"], data["fetched_at"], data, "polymarket.db")
save_snapshot(data["bucket_ts"], data["fetched_at"], data, "../polymarket.db")
print("saved:", data["bucket_ts"], data["fetched_at"])

4
sol抓取.py → polymarket/sol抓取.py
View File

@@ -231,7 +231,7 @@ def save_snapshot(bucket_ts: int, fetched_at: int, raw: Dict[str, Any], db_path:
if __name__ == "__main__":
init_db("polymarket.db")
init_db("../polymarket.db")
while True:
data = fetch_current_market(
@@ -242,5 +242,5 @@ if __name__ == "__main__":
probe_offsets=(0, -900, 900), # 先试当前，再试前一档/后一档
)
save_snapshot(data["bucket_ts"], data["fetched_at"], data, "polymarket.db")
save_snapshot(data["bucket_ts"], data["fetched_at"], data, "../polymarket.db")
print("saved:", data["bucket_ts"], data["fetched_at"])

4
xrp抓取.py → polymarket/xrp抓取.py
View File

@@ -231,7 +231,7 @@ def save_snapshot(bucket_ts: int, fetched_at: int, raw: Dict[str, Any], db_path:
if __name__ == "__main__":
init_db("polymarket.db")
init_db("../polymarket.db")
while True:
data = fetch_current_market(
@@ -242,5 +242,5 @@ if __name__ == "__main__":
probe_offsets=(0, -900, 900), # 先试当前，再试前一档/后一档
)
save_snapshot(data["bucket_ts"], data["fetched_at"], data, "polymarket.db")
save_snapshot(data["bucket_ts"], data["fetched_at"], data, "../polymarket.db")
print("saved:", data["bucket_ts"], data["fetched_at"])

2
text111.py → polymarket/批量抓取.py
View File

@@ -340,7 +340,7 @@ def fetch_crypto_worker(crypto_config: Dict[str, str], db_path: str, stop_event:
def main():
"""主函数：启动多个线程并行抓取所有币种数据"""
db_path = "polymarket.db"
db_path = "../polymarket.db"
init_db(db_path)
stop_event = threading.Event()