通达信量化平台

# 通达信TdxAiData：用批量竞价数据做"上提一字"竞价形态选股

此功能为付费功能，详细使用过程请查看今天的公众号文章 通达信TdxAiData：用批量竞价数据做 (opens new window)

📄 auction_upward_one_line_strategy.py
公众号文中开盘涨幅用的竞价最后一个点的价格，这个代码是把开盘涨幅准确获取的写法

"""竞价上提一字选股。

规则：
1. 真实开盘涨幅（快照 Open / LastClose）在 3% 到 5% 之间，含边界；
2. 竞价末价相对首价上涨超过 0.01%；
3. 只保留至少有 2 个竞价点、且竞价时间在 09:15-09:25 的股票。

运行：
    python auction_upward_one_line_strategy.py

结果保存到当前目录的 upward_one_line_YYYYMMDD.csv。
"""

from __future__ import annotations

import csv
import math
import time
from datetime import datetime
from pathlib import Path

from tdxaidata import tqs


BATCH_SIZE = 300
MIN_OPEN_CHANGE = 3.0
MAX_OPEN_CHANGE = 5.0
MIN_SLOPE = 0.01
FIELDS = ["Time", "Price", "Volume", "TotalNum"]
REFERENCE_FIELDS = ["LastClose", "Open"]
COLUMNS = [
    "stock_code", "first_time", "last_time", "first_price", "final_price",
    "last_close", "open_price", "open_change_pct", "auction_change_pct",
    "slope_pct", "points",
]
OUTPUT = Path(__file__).with_name(
    f"upward_one_line_{datetime.now():%Y%m%d}.csv"
)


def number(value) -> float:
    try:
        result = float(value)
        return result if math.isfinite(result) else 0.0
    except (TypeError, ValueError):
        return 0.0


def auction_points(raw: dict) -> tuple[list[str], list[float]]:
    times = raw.get("Time", [])
    prices = raw.get("Price", [])
    if not isinstance(times, list):
        times = [times]
    if not isinstance(prices, list):
        prices = [prices]

    result_times, result_prices = [], []
    for index, value in enumerate(prices):
        text = str(times[index] if index < len(times) else "")
        digits = "".join(ch for ch in text if ch.isdigit()).zfill(6)[-6:]
        if "091500" <= digits <= "092500":
            price = number(value)
            if price > 0:
                result_times.append(text)
                result_prices.append(price)
    return result_times, result_prices


def stock_pool() -> list[str]:
    result = tqs.get_stock_list(market="5", list_type=0)
    return [
        str(code)
        for code in result or []
        if str(code)[:6].isdigit()
        and str(code).endswith((".SH", ".SZ", ".BJ"))
    ]


def fetch_auction(codes: list[str]) -> dict[str, dict]:
    records = {}
    for start in range(0, len(codes), BATCH_SIZE):
        batch = codes[start:start + BATCH_SIZE]
        for attempt in range(2):
            try:
                result = tqs.get_call_auction_batch(
                    stock_list=batch,
                    field_list=FIELDS,
                    return_df=False,
                )
                if isinstance(result, dict):
                    records.update({
                        code: raw for code, raw in result.items()
                        if isinstance(raw, dict)
                    })
                break
            except Exception:
                if attempt:
                    raise
                time.sleep(1)
        print(f"\r获取竞价：{min(start + BATCH_SIZE, len(codes))}/{len(codes)}",
              end="", flush=True)
    print()
    return records


def fetch_reference(codes: list[str]) -> dict[str, dict]:
    result = {}
    for start in range(0, len(codes), 1000):
        batch = codes[start:start + 1000]
        snapshots = tqs.get_market_snapshot_batch(
            stock_list=batch,
            field_list=REFERENCE_FIELDS,
            return_df=False,
        )
        if isinstance(snapshots, dict):
            for code, raw in snapshots.items():
                if isinstance(raw, dict):
                    result[code] = raw
    return result


def select_stocks(records: dict[str, dict], references: dict[str, dict]) -> list[dict]:
    selected = []
    for code, raw in records.items():
        times, prices = auction_points(raw)
        reference = references.get(code, {})
        close = number(reference.get("LastClose"))
        open_price = number(reference.get("Open"))
        if len(prices) < 2 or close <= 0 or open_price <= 0:
            continue
        # 09:25 前的最后试撮价不一定等于最终成交开盘价。
        open_change = (open_price - close) / close * 100
        auction_change = (prices[-1] - close) / close * 100
        slope = (prices[-1] / prices[0] - 1) * 100
        if MIN_OPEN_CHANGE <= open_change <= MAX_OPEN_CHANGE and slope > MIN_SLOPE:
            selected.append({
                "stock_code": code,
                "first_time": times[0],
                "last_time": times[-1],
                "first_price": prices[0],
                "final_price": prices[-1],
                "last_close": close,
                "open_price": open_price,
                "open_change_pct": round(open_change, 3),
                "auction_change_pct": round(auction_change, 3),
                "slope_pct": round(slope, 4),
                "points": len(prices),
            })

    selected.sort(key=lambda row: row["open_change_pct"], reverse=True)
    return selected


def main() -> None:
    now = datetime.now()
    end = datetime.combine(now.date(), datetime.strptime("09:25", "%H:%M").time())
    if now < end:
        print("等待 09:25 获取完整竞价数据...")
        time.sleep((end - now).total_seconds())

    codes = stock_pool()
    if not codes:
        raise RuntimeError("未获取到股票池，请检查 TdxAiData Token。")

    records = fetch_auction(codes)
    references = fetch_reference(list(records))
    missing_reference = sum(
        number(references.get(code, {}).get("LastClose")) <= 0
        or number(references.get(code, {}).get("Open")) <= 0
        for code in records
    )
    if missing_reference:
        print(f"跳过 {missing_reference} 只：真实开盘价或昨收缺失/无效，不用试撮价替代。")
    selected = select_stocks(records, references)
    with OUTPUT.open("w", newline="", encoding="utf-8-sig") as file:
        writer = csv.DictWriter(file, fieldnames=COLUMNS)
        writer.writeheader()
        writer.writerows(selected)

    print(f"上提一字：{len(selected)} 只")
    for row in selected:
        print(f"{row['stock_code']}  开盘涨幅={row['open_change_pct']:.3f}%  "
              f"竞价斜率={row['slope_pct']:.4f}%  点数={row['points']}  "
              f"开盘价={row['open_price']:.2f}  竞价末价={row['final_price']:.2f}")
    print(f"结果文件：{OUTPUT}")


if __name__ == "__main__":
    main()
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190

📄 auction_upward_one_line_strategy.py
公众号原文的代码 开盘用的是竞价最后一个点计算的

"""竞价上提一字选股。

规则：
1. 竞价最终涨幅在 3% 到 5% 之间，含边界；
2. 竞价末价相对首价上涨超过 0.01%；
3. 只保留至少有 2 个竞价点、且竞价时间在 09:15-09:25 的股票。

运行：
    python auction_upward_one_line_strategy.py

结果保存到当前目录的 upward_one_line_YYYYMMDD.csv。
"""

from __future__ import annotations

import csv
import time
from datetime import datetime
from pathlib import Path

from tdxaidata import tqs


BATCH_SIZE = 300
MIN_OPEN_CHANGE = 3.0
MAX_OPEN_CHANGE = 5.0
MIN_SLOPE = 0.01
FIELDS = ["Time", "Price", "Volume", "TotalNum"]
REFERENCE_FIELDS = ["LastClose"]
OUTPUT = Path(__file__).with_name(
    f"upward_one_line_{datetime.now():%Y%m%d}.csv"
)


def number(value) -> float:
    try:
        return float(value)
    except (TypeError, ValueError):
        return 0.0


def auction_points(raw: dict) -> tuple[list[str], list[float]]:
    times = raw.get("Time", [])
    prices = raw.get("Price", [])
    if not isinstance(times, list):
        times = [times]
    if not isinstance(prices, list):
        prices = [prices]

    result_times, result_prices = [], []
    for index, value in enumerate(prices):
        text = str(times[index] if index < len(times) else "")
        digits = "".join(ch for ch in text if ch.isdigit()).zfill(6)[-6:]
        if "091500" <= digits <= "092500":
            price = number(value)
            if price > 0:
                result_times.append(text)
                result_prices.append(price)
    return result_times, result_prices


def stock_pool() -> list[str]:
    result = tqs.get_stock_list(market="5", list_type=0)
    return [
        str(code)
        for code in result or []
        if str(code)[:6].isdigit()
        and str(code).endswith((".SH", ".SZ", ".BJ"))
    ]


def fetch_auction(codes: list[str]) -> dict[str, dict]:
    records = {}
    for start in range(0, len(codes), BATCH_SIZE):
        batch = codes[start:start + BATCH_SIZE]
        for attempt in range(2):
            try:
                result = tqs.get_call_auction_batch(
                    stock_list=batch,
                    field_list=FIELDS,
                    return_df=False,
                )
                if isinstance(result, dict):
                    records.update({
                        code: raw for code, raw in result.items()
                        if isinstance(raw, dict)
                    })
                break
            except Exception:
                if attempt:
                    raise
                time.sleep(1)
        print(f"\r获取竞价：{min(start + BATCH_SIZE, len(codes))}/{len(codes)}",
              end="", flush=True)
    print()
    return records


def fetch_last_close(codes: list[str]) -> dict[str, float]:
    result = {}
    for start in range(0, len(codes), 1000):
        batch = codes[start:start + 1000]
        snapshots = tqs.get_market_snapshot_batch(
            stock_list=batch,
            field_list=REFERENCE_FIELDS,
            return_df=False,
        )
        if isinstance(snapshots, dict):
            for code, raw in snapshots.items():
                if isinstance(raw, dict):
                    result[code] = number(raw.get("LastClose"))
    return result


def main() -> None:
    now = datetime.now()
    end = datetime.combine(now.date(), datetime.strptime("09:25", "%H:%M").time())
    if now < end:
        print("等待 09:25 获取完整竞价数据...")
        time.sleep((end - now).total_seconds())

    codes = stock_pool()
    if not codes:
        raise RuntimeError("未获取到股票池，请检查 TdxAiData Token。")

    records = fetch_auction(codes)
    last_close = fetch_last_close(list(records))
    selected = []
    for code, raw in records.items():
        times, prices = auction_points(raw)
        close = last_close.get(code, 0.0)
        if len(prices) < 2 or close <= 0:
            continue
        open_change = (prices[-1] / close - 1) * 100
        slope = (prices[-1] / prices[0] - 1) * 100
        if MIN_OPEN_CHANGE <= open_change <= MAX_OPEN_CHANGE and slope > MIN_SLOPE:
            selected.append({
                "stock_code": code,
                "first_time": times[0],
                "last_time": times[-1],
                "first_price": prices[0],
                "final_price": prices[-1],
                "last_close": close,
                "open_change_pct": round(open_change, 3),
                "slope_pct": round(slope, 4),
                "points": len(prices),
            })

    selected.sort(key=lambda row: row["open_change_pct"], reverse=True)
    with OUTPUT.open("w", newline="", encoding="utf-8-sig") as file:
        writer = csv.DictWriter(file, fieldnames=selected[0].keys() if selected else [
            "stock_code", "first_time", "last_time", "first_price",
            "final_price", "last_close", "open_change_pct", "slope_pct", "points",
        ])
        writer.writeheader()
        writer.writerows(selected)

    print(f"上提一字：{len(selected)} 只")
    for row in selected:
        print(f"{row['stock_code']}  开盘涨幅={row['open_change_pct']:.3f}%  "
              f"竞价斜率={row['slope_pct']:.4f}%  点数={row['points']}")
    print(f"结果文件：{OUTPUT}")


if __name__ == "__main__":
    main()
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167

📄 auction_upward_one_line_strategy.py
支持预警和选股到自定义板块 需打开支持TQ的通达信客户端设置此软件目录路径

"""竞价上提一字选股策略。"""
from __future__ import annotations

import pandas as pd
import sys   
sys.path.append('C:/new_tdx_test2025/PYPlugins/user')
from tqcenter import tq
from datetime import datetime

# 1. 初始化
tq.initialize('1009warnbk.py')


import csv
import math
import time
from datetime import datetime
from pathlib import Path

from tdxaidata import tqs


BATCH_SIZE = 300
MIN_OPEN_CHANGE = 3.0
MAX_OPEN_CHANGE = 5.0
MIN_SLOPE = 0.01
FIELDS = ["Time", "Price", "Volume", "TotalNum"]
REFERENCE_FIELDS = ["LastClose", "Open"]
COLUMNS = [
    "stock_code", "first_time", "last_time", "first_price", "final_price",
    "last_close", "open_price", "open_change_pct", "auction_change_pct",
    "slope_pct", "points",
]
OUTPUT = Path(__file__).with_name(
    f"upward_one_line_{datetime.now():%Y%m%d}.csv"
)


def number(value) -> float:
    try:
        result = float(value)
        return result if math.isfinite(result) else 0.0
    except (TypeError, ValueError):
        return 0.0


def auction_points(raw: dict) -> tuple[list[str], list[float]]:
    times = raw.get("Time", [])
    prices = raw.get("Price", [])
    if not isinstance(times, list):
        times = [times]
    if not isinstance(prices, list):
        prices = [prices]

    result_times, result_prices = [], []
    for index, value in enumerate(prices):
        text = str(times[index] if index < len(times) else "")
        digits = "".join(ch for ch in text if ch.isdigit()).zfill(6)[-6:]
        if "091500" <= digits <= "092500":
            price = number(value)
            if price > 0:
                result_times.append(text)
                result_prices.append(price)
    return result_times, result_prices


def stock_pool() -> list[str]:
    result = tqs.get_stock_list(market="5", list_type=0)
    return [
        str(code)
        for code in result or []
        if str(code)[:6].isdigit()
        and str(code).endswith((".SH", ".SZ", ".BJ"))
    ]


def fetch_auction(codes: list[str]) -> dict[str, dict]:
    records = {}
    for start in range(0, len(codes), BATCH_SIZE):
        batch = codes[start:start + BATCH_SIZE]
        for attempt in range(2):
            try:
                result = tqs.get_call_auction_batch(
                    stock_list=batch,
                    field_list=FIELDS,
                    return_df=False,
                )
                if isinstance(result, dict):
                    records.update({
                        code: raw for code, raw in result.items()
                        if isinstance(raw, dict)
                    })
                break
            except Exception:
                if attempt:
                    raise
                time.sleep(1)
        print(
            f"\r获取竞价：{min(start + BATCH_SIZE, len(codes))}/{len(codes)}",
            end="",
            flush=True,
        )
    print()
    return records


def fetch_reference(codes: list[str]) -> dict[str, dict]:
    result = {}
    for start in range(0, len(codes), 1000):
        batch = codes[start:start + 1000]
        snapshots = tqs.get_market_snapshot_batch(
            stock_list=batch,
            field_list=REFERENCE_FIELDS,
            return_df=False,
        )
        if isinstance(snapshots, dict):
            for code, raw in snapshots.items():
                if isinstance(raw, dict):
                    result[code] = raw
    return result


def select_stocks(records: dict[str, dict], references: dict[str, dict]) -> list[dict]:
    selected = []
    for code, raw in records.items():
        times, prices = auction_points(raw)
        reference = references.get(code, {})
        close = number(reference.get("LastClose"))
        open_price = number(reference.get("Open"))
        if len(prices) < 2 or close <= 0 or open_price <= 0:
            continue
        open_change = (open_price - close) / close * 100
        auction_change = (prices[-1] - close) / close * 100
        slope = (prices[-1] / prices[0] - 1) * 100
        if MIN_OPEN_CHANGE <= open_change <= MAX_OPEN_CHANGE and slope > MIN_SLOPE:
            selected.append({
                "stock_code": code,
                "first_time": times[0],
                "last_time": times[-1],
                "first_price": prices[0],
                "final_price": prices[-1],
                "last_close": close,
                "open_price": open_price,
                "open_change_pct": round(open_change, 3),
                "auction_change_pct": round(auction_change, 3),
                "slope_pct": round(slope, 4),
                "points": len(prices),
            })


    selected.sort(key=lambda row: row["open_change_pct"], reverse=True)
    return selected


def main() -> None:
    now = datetime.now()
    end = datetime.combine(now.date(), datetime.strptime("09:25", "%H:%M").time())
    if now < end:
        print("等待 09:25 获取完整竞价数据...")
        time.sleep((end - now).total_seconds())

    codes = stock_pool()
    if not codes:
        raise RuntimeError("未获取到股票池，请检查 TdxAiData Token。")

    records = fetch_auction(codes)
    references = fetch_reference(list(records))
    selected = select_stocks(records, references)



    # 发送预警信号  #发送的是开盘涨幅值
    warn_time = datetime.now().strftime("%Y%m%d%H%M%S")
    stocks = [row["stock_code"] for row in selected]

    if stocks:
        warn_result = tq.send_warn(
            stock_list=stocks,
            time_list=[warn_time] * len(stocks),
            price_list=[str(row["open_price"]) for row in selected],
            close_list=[str(row["last_close"]) for row in selected],
            volum_list=["0"] * len(stocks),
            bs_flag_list=["0"] * len(stocks),
            warn_type_list=["0"] * len(stocks),
            reason_list=["竞价上提一字"] * len(stocks),
            count=len(stocks),
        )
        print(f"预警发送结果：{warn_result}")
    else:
        print("没有符合条件的股票，跳过发送预警。")


    # 创建带日期的自定义板块
    trade_date = datetime.now().strftime("%Y%m%d")
    block_code = f"UP{trade_date}"[:8]
    block_name = f"竞价上提一字_{trade_date}"

    create_result = tq.create_sector(
        block_code=block_code,
        block_name=block_name,
    )
    print(f"创建自定义板块结果：{create_result}")


    # 将选股结果保存到自定义板块
    block_result = tq.send_user_block(
        block_code,
        [row["stock_code"] for row in selected],
        True,
    )
    print(f"保存自定义板块结果：{block_result}")

    tq.close()
        
    with OUTPUT.open("w", newline="", encoding="utf-8-sig") as file:
        writer = csv.DictWriter(file, fieldnames=COLUMNS)
        writer.writeheader()
        writer.writerows(selected)

    print(f"上提一字：{len(selected)} 只")
    for row in selected:
        print(
            f"{row['stock_code']} 开盘涨幅={row['open_change_pct']:.3f}% "
            f"竞价斜率={row['slope_pct']:.4f}% 点数={row['points']} "
            f"开盘价={row['open_price']:.2f} 竞价末价={row['final_price']:.2f}"
        )
    print(f"结果文件：{OUTPUT}")


if __name__ == "__main__":
    main()
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231

← 常见问题 →