同花顺添加删除自选股演示示例

import requests
import re
import json
import time
import pandas as pd
import sys
from pathlib import Path
# 获取项目根目录路径（兼容不同入口）
# BASE_DIR = Path(__file__).resolve().parent.parent
# sys.path.append(str(BASE_DIR))  # 将根目录加入搜索路径
#
# from config.settings import Config
class TonghuashunAPI:
    def __init__(self, cookie):
        """
        初始化同花顺API接口
        :param cookie: 同花顺登录后的Cookie字符串
        """
        self.cookie = cookie
        self.headers = {
            'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/125.0.0.0 Safari/537.36',
            'Referer': 'https://upass.10jqka.com.cn/login',
            'Cookie': cookie,
            'DNT': '1'
        }
        self.base_url = "https://t.10jqka.com.cn"
    def get_self_stock(self):
        """
        获取同花顺自选股列表（带市场代码）
        返回格式示例：selfStock({"errorCode":0,"errorMsg":"","result":[{"code":"600519","marketid":"17"}],"isT":true})
        """
        try:
            # 生成防缓存时间戳（精确到毫秒）
            timestamp = int(time.time() * 1000)
            url = f'{self.base_url}/newcircle/group/getSelfStockWithMarket/?callback=selfStock&_={timestamp}'
            response = requests.get(url, headers=self.headers, timeout=10)
            response.raise_for_status()  # 检查HTTP错误
            return response.text
        except Exception as e:
            print(f"获取自选股失败: {e}")
            return None
    def add_stock(self, code, marketid):
        """
        添加自选股
        :param code: 股票代码，如 "600519"
        :param marketid: 市场代码，如 "17"（沪市）
        :return: 操作结果
        """
        try:
            timestamp = int(time.time() * 1000)
            url = f'{self.base_url}/newcircle/group/modifySelfStock/?callback=modifyStock&op=add&&stockcode={code}&_={timestamp}'
            response = requests.get(url, headers=self.headers, timeout=10)
            response.raise_for_status()
            return self.parse_jsonp(response.text, 'modifyStock')
        except Exception as e:
            print(f"添加自选股失败: {e}")
            return {'errorCode': -1, 'errorMsg': str(e)}
    def delete_stock(self, code, marketid):
        """
        删除自选股
        :param code: 股票代码，如 "600519"
        :param marketid: 市场代码，如 "17"（沪市）
        :return: 操作结果
        """
        try:
            timestamp = int(time.time() * 1000)
            url = f'{self.base_url}/newcircle/group/modifySelfStock?op=del&stockcode={code}&marketid={marketid}&_={timestamp}'
            response = requests.get(url, headers=self.headers, timeout=10)
            response.raise_for_status()
            return self.parse_jsonp(response.text, 'modifyStock')
        except Exception as e:
            print(f"删除自选股失败: {e}")
            return {'errorCode': -1, 'errorMsg': str(e)}
    def batch_delete_stocks(self, stock_list):
        """
        批量删除自选股
        :param stock_list: 股票列表，格式 [{"code": "600519", "marketid": "17"}, ...]
        :return: 批量操作结果
        """
        results = []
        for stock in stock_list:
            result = self.delete_stock(stock['code'], stock['marketid'])
            results.append({
                'code': stock['code'],
                'marketid': stock['marketid'],
                'result': result
            })
            # 添加短暂延迟，避免请求过于频繁
            time.sleep(0.1)
        return results
    def parse_jsonp(self, jsonp_str, callback_name='selfStock'):
        """
        解析JSONP响应为Python字典
        :param jsonp_str: JSONP格式的字符串
        :param callback_name: 回调函数名
        """
        try:
            # 方法1：使用正则提取JSON部分
            pattern = re.compile(fr'{re.escape(callback_name)}\((.*)\);?', re.DOTALL)
            match = pattern.search(jsonp_str)
            if match:
                json_str = match.group(1)
                return json.loads(json_str)
            else:
                # 方法2：直接去除回调函数名和括号
                stripped = jsonp_str.replace(f"{callback_name}(", "").replace(");", "")
                return json.loads(stripped)
        except Exception as e:
            print(f"JSONP解析失败: {e}")
            return {'errorCode': -1, 'errorMsg': f'JSONP解析失败: {e}'}
    def convert_to_dataframe(self, data):
        """
        将解析后的数据转换为DataFrame
        """
        if data and data.get('errorCode') == 0:
            return pd.DataFrame(data['result'])
        return pd.DataFrame()
    def check_login_status(self):
        """
        检查登录状态
        :return: True表示已登录，False表示未登录
        """
        data = self.get_self_stock()
        if data:
            parsed = self.parse_jsonp(data)
            return parsed.get('errorCode') == 0
        return False
    def get_market_name(self, marketid):
        """
        根据市场代码获取市场名称
        """
        market_map = {
            '17': '沪市',
            '33': '深市',
            '151': '北交所',
            '87': '北交所',  # 可能有不同的代码表示同一市场
            '1': '美股',
            '2': '港股'
        }
        return market_map.get(marketid, f'未知市场({marketid})')
# 使用示例
if __name__ == "__main__":
    # 替换为您的实际Cookie（通过浏览器开发者工具获取）
    #USER_COOKIE = Config.TONGHUASHUN_COOKIE
    USER_COOKIE = "添加你的同花顺cookie"

    # 创建API实例
    ths = TonghuashunAPI(cookie=USER_COOKIE)
    # 检查登录状态
    if not ths.check_login_status():
        print("用户未登录，请检查Cookie有效性")
        exit(1)
    print("用户已登录，开始操作自选股...")
    # 1. 获取当前自选股列表
    print("\n1. 获取当前自选股列表:")
    jsonp_response = ths.get_self_stock()
    if jsonp_response:
        parsed_data = ths.parse_jsonp(jsonp_response)
        df_stocks = ths.convert_to_dataframe(parsed_data)
        if not df_stocks.empty:
            print("当前自选股:")
            def add_market_suffix(row):
                code = row['code']
                market = row['marketid']
                market_name = ths.get_market_name(market)
                if market == '17':  # 沪市
                    return f"{code}.SH", market_name
                elif market == '33':  # 深市
                    return f"{code}.SZ", market_name
                elif market in ['151', '87']:  # 北交所
                    return f"{code}.BJ", market_name
                return code, market_name
            df_stocks[['full_code', 'market_name']] = df_stocks.apply(
                lambda row: pd.Series(add_market_suffix(row)), axis=1
            )
            print(df_stocks[['code', 'marketid', 'market_name', 'full_code']])
        else:
            print("当前没有自选股")
    # 2. 添加自选股示例
    print("\n2. 添加自选股示例:")
    # 添加贵州茅台（沪市）
    add_result = ths.add_stock("600519", "17")
    if add_result.get('errorCode') == 0:
        print("添加贵州茅台(600519.SH)成功")
    else:
        print(f"添加失败: {add_result.get('errorMsg', '未知错误')}")
    # 添加宁德时代（深市）
    add_result = ths.add_stock("300750", "33")
    if add_result.get('errorCode') == 0:
        print("添加宁德时代(300750.SZ)成功")
    else:
        print(f"添加失败: {add_result.get('errorMsg', '未知错误')}")
    # 3. 删除自选股示例
    print("\n3. 删除自选股示例:")
    # 删除贵州茅台
    delete_result = ths.delete_stock("600519", "17")
    if delete_result.get('errorCode') == 0:
        print("删除贵州茅台(600519.SH)成功")
    else:
        print(f"删除失败: {delete_result.get('errorMsg', '未知错误')}")
    # 4. 批量操作示例
    print("\n4. 批量操作示例:")
    stocks_to_delete = [
        {"code": "300750", "marketid": "33"},
        # 可以添加更多要删除的股票
    ]
    batch_results = ths.batch_delete_stocks(stocks_to_delete)
    for result in batch_results:
        stock_info = f"{result['code']}(市场:{result['marketid']})"
        if result['result'].get('errorCode') == 0:
            print(f"删除 {stock_info} 成功")
        else:
            print(f"删除 {stock_info} 失败: {result['result'].get('errorMsg', '未知错误')}")
    # 5. 最终的自选股列表
    print("\n5. 最终的自选股列表:")
    final_response = ths.get_self_stock()
    if final_response:
        final_data = ths.parse_jsonp(final_response)
        final_df = ths.convert_to_dataframe(final_data)
        if not final_df.empty:
            final_df[['full_code', 'market_name']] = final_df.apply(
                lambda row: pd.Series(add_market_suffix(row)), axis=1
            )
            print(final_df[['code', 'marketid', 'market_name', 'full_code']])
        else:
            print("当前没有自选股")