提交代码

polymarket 抓取价格.py

@@ -0,0 +1,78 @@
+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-1766912400", decimals=0))

test1.py

@@ -1,31 +1,32 @@
 import requests
-import uuid
+import json
 
-session = requests.Session()
+GAMMA = "https://gamma-api.polymarket.com"
 
-headers = {
-    "User-Agent": "ShopeeApp/3.54.2 (Android 13; SM-G991B)",  # 新版本 UA
-    "Accept": "application/json",
-    "Accept-Language": "id-ID",
-    "X-Requested-With": "com.shopee.id",
-    "Referer": "https://shopee.co.id/",
-    "Connection": "keep-alive",
-    "X-Shopee-Device-Id": str(uuid.uuid4()),  # 每次请求随机 device_id
-}
+def parse_jsonish_list(v):
+    if isinstance(v, list):
+        return v
+    if isinstance(v, str):
+        return json.loads(v)
+    return []
 
-def get_item(itemid, shopid):
-    url = "https://shopee.co.id/api/v4/item/get"
-    params = {"itemid": itemid, "shopid": shopid}
-    r = session.get(url, headers=headers, params=params, timeout=10)
+def get_tokens_from_slug(slug: str):
+    r = requests.get(f"{GAMMA}/markets/slug/{slug}", timeout=20)
     r.raise_for_status()
-    return r.json()
+    m = r.json()
+
+    outcomes = parse_jsonish_list(m.get("outcomes"))
+    token_ids = parse_jsonish_list(m.get("clobTokenIds"))
+
+    mapping = dict(zip(outcomes, token_ids))
+
+    return {
+        "question": m.get("question"),
+        "market_id": m.get("id"),
+        "outcomes": outcomes,
+        "token_map": mapping,  # {"Up": "...", "Down": "..."}
+    }
 
 if __name__ == "__main__":
-    data = get_item(4420309814, 10115139)
-    item = data.get("data", {}).get("item")
-    if item:
-        print("商品名:", item["name"])
-        print("价格:", item["price"] // 100000)
-        print("库存:", item["stock"])
-    else:
-        print("无数据 / 被风控")
+    info = get_tokens_from_slug("eth-updown-15m-1766912400")
+    print(info)

test111.py

@@ -1,64 +0,0 @@
-import time
-from loguru import *
-from urllib.parse import urlparse, parse_qs
-
-import pandas as pd
-from DrissionPage import *
-
-if __name__ == '__main__':
-
-    co = ChromiumOptions()
-    co.set_local_port(9999)
-
-    co.set_local_port(9999)
-    co.headless(False)  # 先用有头模式测试，看是否还能触发
-    co.incognito(False)  # 避免无痕
-    co.set_argument('--disable-blink-features=AutomationControlled')
-    co.set_argument('--no-sandbox')
-    co.set_argument('--disable-infobars')
-    co.set_argument('--disable-extensions')
-    co.set_argument('--start-maximized')
-    co.set_user_agent("Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36")  # 真实 UA
-
-    page = ChromiumPage(addr_or_opts=co)
-
-    # 读取 Excel 文件
-    excel_file = pd.ExcelFile('sample_items_1000_api.xlsx')
-
-    # 获取指定工作表中的数据
-    df = excel_file.parse('in')
-
-    # 读取第一列数据
-    first_column = df.iloc[:, 0]
-
-    # 逐行遍历第一列数据
-    n = 0
-    n1 = 0
-    for value in first_column:
-        print(value)
-
-        # 解析 URL
-        parsed_url = urlparse(value)
-
-        # 提取查询参数部分
-        query_params = parsed_url.query
-
-        # 解析查询参数为字典
-        param_dict = parse_qs(query_params)
-
-        print(param_dict['shop_id'][0])
-        print(param_dict['item_id'][0])
-
-        # a = time.time()
-        # tab = page.new_tab()
-        # tab.listen.start("shopee.tw/api/v4/pdp/get_pc")
-        # tab.get(f"https://shopee.tw/product/{param_dict['shop_id'][0]}/{param_dict['item_id'][0]}")
-        # res = tab.listen.wait(timeout=15)
-        # print(res.response.body)
-        # if res.response.body.get("data", {}).get("item", {}):
-        #     n += 1
-        #     n1 += time.time() - a
-        #     # https: // shopee.tw / api / v4 / pdp / get_pc?item_id = 22577587881 & shop_id = 1014505717 & tz_offset_minutes = 480 & detail_level = 0 &            logger.info(f"成功第{n}个，耗时{time.time() - a}秒")
-        #     logger.success(f"成功第{n}个，一共耗时{n1}秒")
-        #
-        # tab.close()