91 lines
3.0 KiB
Python
91 lines
3.0 KiB
Python
|
||
import os
|
||
import requests
|
||
import json
|
||
import time
|
||
|
||
class IFindClient:
|
||
def __init__(self, refresh_token: str):
|
||
self.refresh_token = refresh_token
|
||
self.access_token = None
|
||
self.token_expiry = 0
|
||
self.base_url = "https://quantapi.51ifind.com/api/v1"
|
||
|
||
def _get_access_token(self):
|
||
"""获取当前的 access_token，如果过期则刷新"""
|
||
# 简单判断，手册说有效期 7 天，我们这里如果没 token 就去换一个
|
||
if self.access_token and time.time() < self.token_expiry:
|
||
return self.access_token
|
||
|
||
url = f"{self.base_url}/get_access_token"
|
||
headers = {
|
||
"Content-Type": "application/json",
|
||
"refresh_token": self.refresh_token
|
||
}
|
||
|
||
try:
|
||
response = requests.post(url, headers=headers)
|
||
data = response.json()
|
||
if data.get("errorcode") == 0:
|
||
self.access_token = data["data"]["access_token"]
|
||
# 设个保守的有效期，比如 6 天
|
||
self.token_expiry = time.time() + (6 * 24 * 3600)
|
||
return self.access_token
|
||
else:
|
||
raise Exception(f"iFinD login failed: {data.get('errmsg')}")
|
||
except Exception as e:
|
||
print(f"Error getting iFinD access token: {e}")
|
||
return None
|
||
|
||
def post(self, endpoint, params):
|
||
token = self._get_access_token()
|
||
if not token:
|
||
return None
|
||
|
||
url = f"{self.base_url}/{endpoint}"
|
||
headers = {
|
||
"Content-Type": "application/json",
|
||
"access_token": token
|
||
}
|
||
|
||
try:
|
||
response = requests.post(url, headers=headers, json=params)
|
||
return response.json()
|
||
except Exception as e:
|
||
print(f"iFinD request error on {endpoint}: {e}")
|
||
return None
|
||
|
||
if __name__ == "__main__":
|
||
# 简单测试代码
|
||
from dotenv import load_dotenv
|
||
load_dotenv()
|
||
|
||
token = os.getenv("IFIND_REFRESH_TOKEN")
|
||
if token:
|
||
client = IFindClient(token)
|
||
|
||
code = "7203.T"
|
||
print(f"--- Testing date_sequence Global indicators for {code} ---")
|
||
params = {
|
||
"codes": code,
|
||
"startdate": "2024-12-01",
|
||
"enddate": "2024-12-10",
|
||
"functionpara": {"Interval": "D", "Days": "Alldays", "Fill": "Blank"},
|
||
"indipara": [
|
||
{"indicator": "pe_ttm", "indiparams": ["1", "BB"]},
|
||
{"indicator": "pb", "indiparams": ["1", "BB"]},
|
||
{"indicator": "total_mv", "indiparams": ["1", "BB"]}
|
||
]
|
||
}
|
||
res = client.post("date_sequence", params)
|
||
if res and res.get('errorcode') == 0:
|
||
tables = res.get('tables', [])
|
||
if tables:
|
||
print(json.dumps(tables[0]['table'], indent=2))
|
||
else:
|
||
print("Empty tables")
|
||
else:
|
||
print("Failed")
|
||
else:
|
||
print("No IFIND_REFRESH_TOKEN found in .env")
|