eta_model/WTI/api.py

import random
import string
import time
import base64
from hashlib import sha256
from hmac import HMAC

import http.client
import requests
import pandas as pd 

APPID = "tubmafwrzhpgfiuf"
SECRET = "eotpcqbvhycdshwscqnytiwzbgonposs"


def generate_nonce(length=32):
    """Generate a random nonce."""
    return ''.join(random.choices(string.ascii_letters + string.digits, k=length))


def get_timestamp():
    """Get the current timestamp."""
    return int(time.time())


def build_sign_str(appid, nonce, timestamp):
    """Build the string to be signed."""
    return f'appid={appid}&nonce={nonce}&timestamp={timestamp}'


def calculate_signature(secret, message):
    """Calculate the HMAC SHA-256 signature."""
    return base64.urlsafe_b64encode(HMAC(secret.encode('utf-8'), message.encode('utf-8'), sha256).digest()).decode('utf-8')


def fetch_indicator_details(indicator_id):
    """Fetch the details for a specific indicator ID."""
    nonce = generate_nonce()
    timestamp = get_timestamp()
    sign_str = build_sign_str(APPID, nonce, timestamp)
    signature = calculate_signature(SECRET, sign_str)

    headers = {
        'nonce': nonce,
        'timestamp': str(timestamp),
        'appid': APPID,
        'signature': signature,
        'Accept': "*/*",
        'Accept-Encoding': "gzip, deflate, br",
        'User-Agent': "PostmanRuntime-ApipostRuntime/1.1.0",
        'Connection': "keep-alive",
    }

    url = f"https://etahub.hzinsights.com/v1/edb/data?EdbCode={indicator_id}"
    response = requests.get(url, headers=headers)

    if response.status_code == 200:
        return response.json().get('Data')
    else:
        print(
            f"Failed to fetch data for ID {indicator_id}, status code: {response.status_code}")
        return None

def fetch_indicator_name(indicator_id):
    """Fetch the name for a specific indicator ID."""
    nonce = generate_nonce()
    timestamp = get_timestamp()
    sign_str = build_sign_str(APPID, nonce, timestamp)
    signature = calculate_signature(SECRET, sign_str)

    headers = {
        'nonce': nonce,
        'timestamp': str(timestamp),
        'appid': APPID,
        'signature': signature,
        'Accept': "*/*",
        'Accept-Encoding': "gzip, deflate, br",
        'User-Agent': "PostmanRuntime-ApipostRuntime/1.1.0",
        'Connection': "keep-alive",
    }

    url = f"https://etahub.hzinsights.com/v1/edb/detail?EdbCode={indicator_id}"
    response = requests.get(url, headers=headers)

    if response.status_code == 200:
        return response.json().get('Data').get('EdbName')
    else:
        print(
            f"Failed to fetch data for ID {indicator_id}, status code: {response.status_code}")
        return None

def fetch_data_by_indicators(indicator_ids, output_path=None):
    """
    根据提供的indicator IDs获取数据并返回DataFrame
    
    参数:
    indicator_ids (list): 指标ID列表
    output_path (str, optional): 如果提供，将结果保存为Excel文件的路径
    
    返回:
    pandas.DataFrame: 包含所有指标数据的DataFrame
    """
    # Dictionary to store DataFrames for each indicator
    data_frames = {}

    for indicator_id in indicator_ids:
        data = fetch_indicator_details(indicator_id)
        if data:
            df = pd.DataFrame(data)
            df['DataTime'] = pd.to_datetime(df['DataTime'])
            df.set_index('DataTime', inplace=True)
            df.sort_index(inplace=True)
            df = df[['Value']].rename(columns={'Value': fetch_indicator_name(indicator_id)})
            data_frames[indicator_id] = df

    # Concatenate all DataFrames along the columns
    if data_frames:
        result_df = pd.concat(data_frames.values(), axis=1)
        
        if output_path:
            result_df.to_excel(output_path)
            print(f"Data saved successfully as '{output_path}'")
            
        return result_df
    return None


'''
# 示例用法
if __name__ == "__main__":
    example_indicators = ["RBWTICKMc1", "C2406121350446455"]
    df = fetch_data_by_indicators(example_indicators, "data_input/RBOB.xlsx")
    print(df.info())
'''