-
Notifications
You must be signed in to change notification settings - Fork 1
Expand file tree
/
Copy pathserver.py
More file actions
9298 lines (8597 loc) · 449 KB
/
Copy pathserver.py
File metadata and controls
9298 lines (8597 loc) · 449 KB
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
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967
968
969
970
971
972
973
974
975
976
977
978
979
980
981
982
983
984
985
986
987
988
989
990
991
992
993
994
995
996
997
998
999
1000
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
深脉 DeepPulse · 金融工作台 — 本地数据服务
====================================================
由 DeepSeek 为自己打造的「身体」的后端神经中枢。
· 零第三方依赖(仅 Python 标准库)
· 行情数据源:通达信 TQ-Local(可选本地增强)→ 东方财富 → 腾讯备援
· 官方披露源:巨潮资讯结构化公告;上交所、深交所、证监会作为一级查验入口
· 内置:情绪周期策略引擎(emotion.py)、每日情绪快照记忆(data/history.json)
· 内置:多级缓存 + 上游限频,礼貌访问上游
启动:python server.py
(自动在 8971~8980 中选择可用端口,并将实际端口写入 data/port.txt)
"""
import json
import hashlib
import hmac
import io
import importlib
import importlib.util
import os
import re
import secrets
import socket
import struct
import sys
import threading
import time
import urllib.request
import urllib.parse
import urllib.error
import zipfile
from datetime import datetime, timezone, timedelta, date as _date
from http.client import RemoteDisconnected
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
try:
import tdx_local as tdx_local_api
except Exception:
tdx_local_api = None
try:
from event_impact import build_event_impact, MODEL_VERSION as EVENT_IMPACT_MODEL_VERSION
except Exception:
build_event_impact = None
EVENT_IMPACT_MODEL_VERSION = 'event-impact-unavailable'
try:
from attention_triage import (build_attention_triage,
attention_relevance_scope,
classify_attention_topic,
TOPIC_TAXONOMY_FINGERPRINT,
MODEL_VERSION as ATTENTION_TRIAGE_MODEL_VERSION)
except Exception:
build_attention_triage = None
attention_relevance_scope = classify_attention_topic = None
TOPIC_TAXONOMY_FINGERPRINT = 'attention-topic-unavailable'
ATTENTION_TRIAGE_MODEL_VERSION = 'attention-triage-unavailable'
try:
from observation_rules import (parse_intent as parse_observation_intent,
normalize_draft as normalize_observation_draft,
evaluate as evaluate_observation_rule,
describe_clause as describe_observation_clause,
SIGNALS as OBSERVATION_SIGNALS,
MODEL_VERSION as OBSERVATION_RULES_MODEL_VERSION)
except Exception:
parse_observation_intent = normalize_observation_draft = None
evaluate_observation_rule = describe_observation_clause = None
OBSERVATION_SIGNALS = {}
OBSERVATION_RULES_MODEL_VERSION = 'observation-rules-unavailable'
try:
from research_hypothesis import (create_hypothesis, review_hypothesis,
hypothesis_snapshot,
MODEL_VERSION as HYPOTHESIS_MODEL_VERSION)
except Exception:
create_hypothesis = review_hypothesis = hypothesis_snapshot = None
HYPOTHESIS_MODEL_VERSION = 'research-hypothesis-unavailable'
try:
from research_memory import (build_snapshot as build_research_memory_snapshot,
normalize_preferences as normalize_research_memory_preferences,
MODEL_VERSION as RESEARCH_MEMORY_MODEL_VERSION)
except Exception:
build_research_memory_snapshot = normalize_research_memory_preferences = None
RESEARCH_MEMORY_MODEL_VERSION = 'research-memory-unavailable'
try:
from research_workflow import (attach_workflow_lineage, create_workflow, mutate_workflow, preview_workflow,
record_run as record_workflow_run, workflow_snapshot,
MODEL_VERSION as RESEARCH_WORKFLOW_MODEL_VERSION)
except Exception:
attach_workflow_lineage = create_workflow = mutate_workflow = preview_workflow = None
record_workflow_run = workflow_snapshot = None
RESEARCH_WORKFLOW_MODEL_VERSION = 'research-workflow-unavailable'
try:
from research_watch import (confirm_watch as confirm_research_watch,
is_due as research_watch_is_due,
material_change as research_watch_material_change,
method_fingerprint as research_watch_method_fingerprint,
next_check_at as research_watch_next_check,
preview_watch as preview_research_watch,
watch_state as research_watch_state,
MODEL_VERSION as RESEARCH_WATCH_MODEL_VERSION)
except Exception:
confirm_research_watch = research_watch_is_due = research_watch_material_change = None
research_watch_method_fingerprint = research_watch_next_check = preview_research_watch = None
research_watch_state = None
RESEARCH_WATCH_MODEL_VERSION = 'research-watch-unavailable'
try:
from ai_research_duty import (build_messages as build_ai_research_messages,
confirm_delegation as confirm_ai_research_delegation,
create_job as create_ai_research_job,
delegation_state as ai_research_delegation_state,
parse_draft as parse_ai_research_draft,
preview_delegation as preview_ai_research_delegation,
provider_status as ai_research_provider_status,
MODEL_VERSION as AI_RESEARCH_DUTY_MODEL_VERSION)
except Exception:
build_ai_research_messages = confirm_ai_research_delegation = None
create_ai_research_job = ai_research_delegation_state = None
parse_ai_research_draft = preview_ai_research_delegation = None
ai_research_provider_status = None
AI_RESEARCH_DUTY_MODEL_VERSION = 'ai-research-duty-unavailable'
try:
from ai_provider import (candidate as ai_provider_candidate,
config_fingerprint as ai_provider_config_fingerprint,
public_status as ai_provider_public_status,
test_preview as ai_provider_test_preview,
validate_confirmation as validate_ai_provider_confirmation,
MODEL_VERSION as AI_PROVIDER_MODEL_VERSION,
SCHEMA_VERSION as AI_PROVIDER_SCHEMA_VERSION)
except Exception:
ai_provider_candidate = ai_provider_config_fingerprint = None
ai_provider_public_status = ai_provider_test_preview = None
validate_ai_provider_confirmation = None
AI_PROVIDER_MODEL_VERSION = 'ai-provider-unavailable'
AI_PROVIDER_SCHEMA_VERSION = 'research-draft-schema-unavailable'
try:
from ai_service_management import (
build_status as build_ai_service_status,
normalize_preferences as normalize_ai_service_preferences,
preview_preferences as preview_ai_service_preferences,
validate_plan as validate_ai_service_plan,
MODEL_VERSION as AI_SERVICE_MANAGEMENT_MODEL_VERSION)
except Exception:
build_ai_service_status = normalize_ai_service_preferences = None
preview_ai_service_preferences = validate_ai_service_plan = None
AI_SERVICE_MANAGEMENT_MODEL_VERSION = 'ai-service-management-unavailable'
try:
from research_suggestions import (build_snapshot as build_research_suggestion_snapshot,
draft_fingerprint as research_suggestion_draft_fingerprint,
mutate_item as mutate_research_suggestion_item,
MODEL_VERSION as RESEARCH_SUGGESTION_MODEL_VERSION)
except Exception:
build_research_suggestion_snapshot = research_suggestion_draft_fingerprint = None
mutate_research_suggestion_item = None
RESEARCH_SUGGESTION_MODEL_VERSION = 'research-suggestions-unavailable'
try:
from akshare_research import (build_snapshot as build_akshare_research_snapshot,
unloaded_snapshot as unloaded_akshare_research_snapshot,
normalize_pack_ids as normalize_akshare_pack_ids,
pack_catalog as akshare_pack_catalog,
MODEL_VERSION as AKSHARE_RESEARCH_MODEL_VERSION)
except Exception:
build_akshare_research_snapshot = unloaded_akshare_research_snapshot = None
normalize_akshare_pack_ids = akshare_pack_catalog = None
AKSHARE_RESEARCH_MODEL_VERSION = 'akshare-research-unavailable'
try:
from hypothesis_evidence import (collect_candidate_evidence,
MODEL_VERSION as HYPOTHESIS_EVIDENCE_MODEL_VERSION)
except Exception:
collect_candidate_evidence = None
HYPOTHESIS_EVIDENCE_MODEL_VERSION = 'hypothesis-evidence-unavailable'
_akshare_module = None
_akshare_error = None
# ---------------------------------------------------------------- 基础配置
BASE = os.path.dirname(os.path.abspath(__file__))
WEB = os.path.join(BASE, 'web')
DATA = os.path.join(BASE, 'data')
HISTORY_FILE = os.path.join(DATA, 'history.json')
PROFILE_FILE = os.path.join(DATA, 'profile.json')
SECTOR_HISTORY_FILE = os.path.join(DATA, 'sector_history.json')
DEVICE_CONFIG_FILE = os.path.join(DATA, 'device_config.json')
PORT_FILE = os.path.join(DATA, 'port.txt')
LOG_FILE = os.path.join(DATA, 'server.log')
DIAGNOSTICS_HISTORY_FILE = os.path.join(DATA, 'diagnostics_history.json')
os.makedirs(DATA, exist_ok=True)
BJC = timezone(timedelta(hours=8)) # 北京时间(A股时区)
UA_HEADERS = {
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 '
'(KHTML, like Gecko) Chrome/126.0 Safari/537.36',
'Accept': '*/*',
'Referer': 'https://quote.eastmoney.com/',
}
EM_UT = '7eea3edcaed734bea9cbfc24409ed989' # 东财公开 token
TDX_ENABLED = os.environ.get('DEEPPULSE_TDX_ENABLED', '1').strip().lower() not in ('0', 'false', 'off')
TDX_HOST = '127.0.0.1:17709'
VERSION = '1.42.5'
_desktop_heartbeat_lock = threading.Lock()
_desktop_heartbeat = {
'last_seen': None,
'app_version': None,
'product_version': None,
'service_ownership': None,
'process_lifetime_protected': None,
}
_diagnostics_history_lock = threading.Lock()
try:
from emotion import (compute_emotion, DEFAULT_WEIGHTS, load_weights, # 情绪引擎
save_weights, INDICATORS)
except Exception: # 引擎不可用时降级
compute_emotion = None
DEFAULT_WEIGHTS = {}
INDICATORS = []
load_weights = lambda: {}
save_weights = lambda w: {}
# ---------------------------------------------------------------- 工具函数
class DeepPulseHTTPServer(ThreadingHTTPServer):
"""Avoid Windows' permissive HTTPServer port sharing semantics."""
allow_reuse_address = False
def port_is_listening(host, port, timeout=0.2):
"""Return True when another local service already accepts connections."""
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
try:
sock.settimeout(timeout)
return sock.connect_ex((host, port)) == 0
finally:
sock.close()
def log(msg):
try:
with open(LOG_FILE, 'a', encoding='utf-8') as f:
f.write('[%s] %s\n' % (now_bj().strftime('%Y-%m-%d %H:%M:%S'), msg))
except Exception:
pass
def now_bj():
return datetime.now(BJC)
def today_str():
return now_bj().strftime('%Y-%m-%d')
# 上游请求限频:全局串行(同一时刻只发一个上游请求,至少间隔 0.2s)
# ——实测并行突发会触发上游限流丢包,串行礼貌访问最稳(2026-08-15 夜测结论)
_host_lock = threading.Lock()
_last_req = {}
_source_lock = threading.Lock()
_source_stats = {}
SOURCE_CATALOG = [
{
'id': 'tdx_local', 'name': '通达信 TQ-Local', 'tier': 'local',
'role': '本机实时行情、K线与市场情绪统计交叉验证(严格只读)',
'homepage': 'https://help.tdx.com.cn/quant/',
'hosts': [TDX_HOST], 'mode': 'local',
},
{
'id': 'cninfo', 'name': '巨潮资讯', 'tier': 'official',
'role': '上市公司法定信息披露与公告原文',
'homepage': 'https://www.cninfo.com.cn/new/index',
'hosts': ['www.cninfo.com.cn'], 'mode': 'live',
},
{
'id': 'sse', 'name': '上海证券交易所', 'tier': 'official',
'role': '沪市上市公司公告查验入口',
'homepage': 'https://www.sse.com.cn/disclosure/listedinfo/announcement/',
'hosts': ['www.sse.com.cn'], 'mode': 'reference',
},
{
'id': 'szse', 'name': '深圳证券交易所', 'tier': 'official',
'role': '深市上市公司公告查验入口',
'homepage': 'https://www.szse.cn/disclosure/notice/company/index.html',
'hosts': ['www.szse.cn'], 'mode': 'reference',
},
{
'id': 'csrc', 'name': '中国证监会', 'tier': 'official',
'role': '监管政策、行政许可与监管信息查验入口',
'homepage': 'https://www.csrc.gov.cn/',
'hosts': ['www.csrc.gov.cn'], 'mode': 'reference',
},
{
'id': 'eastmoney', 'name': '东方财富', 'tier': 'market',
'role': '实时行情、K线、资金流、涨跌停池与市场快讯',
'homepage': 'https://quote.eastmoney.com/',
'hosts': ['push2.eastmoney.com', 'push2delay.eastmoney.com',
'push2his.eastmoney.com', 'push2ex.eastmoney.com',
'newsapi.eastmoney.com'],
'mode': 'live',
},
{
'id': 'tencent', 'name': '腾讯行情', 'tier': 'market',
'role': '个股行情与K线备援',
'homepage': 'https://gu.qq.com/',
'hosts': ['qt.gtimg.cn', 'web.ifzq.gtimg.cn'], 'mode': 'fallback',
},
{
'id': 'akshare', 'name': 'AKShare', 'tier': 'enrichment',
'role': '交易日历、宏观与跨市场公开数据补充;不作为实时行情或官方披露主源',
'homepage': 'https://akshare.akfamily.xyz/',
'hosts': [], 'mode': 'optional_local',
},
]
def load_akshare():
global _akshare_module, _akshare_error
if _akshare_module is not None:
return _akshare_module
try:
_akshare_module = importlib.import_module('akshare')
_akshare_error = None
return _akshare_module
except Exception as exc:
_akshare_error = str(exc)[:200]
return None
def akshare_trade_dates():
module = load_akshare()
if module is None or not hasattr(module, 'tool_trade_date_hist_sina'):
raise RuntimeError('AKShare 交易日历接口不可用')
started = time.monotonic()
try:
frame = module.tool_trade_date_hist_sina()
values = frame['trade_date'].tolist()
dates = {str(value)[:10] for value in values}
_record_source('akshare:tool_trade_date_hist_sina', True,
(time.monotonic() - started) * 1000)
return dates
except Exception as exc:
_record_source('akshare:tool_trade_date_hist_sina', False,
(time.monotonic() - started) * 1000, exc)
raise
def market_calendar_info(current=None):
value = (current or now_bj()).astimezone(BJC)
day = value.strftime('%Y-%m-%d')
if value.weekday() >= 5:
return {'date': day, 'is_trade_date': False, 'confirmed': True,
'basis': 'weekend', 'source': 'system-calendar'}
if importlib.util.find_spec('akshare') is not None:
try:
dates = cached('akshare_trade_calendar', 12 * 3600, akshare_trade_dates)
return {'date': day, 'is_trade_date': day in dates, 'confirmed': True,
'basis': 'AKShare 交易日历', 'source': 'akshare'}
except Exception as exc:
return {'date': day, 'is_trade_date': True, 'confirmed': False,
'basis': '工作日降级判断', 'source': 'system-calendar',
'error': str(exc)[:160]}
return {'date': day, 'is_trade_date': True, 'confirmed': False,
'basis': '工作日降级判断', 'source': 'system-calendar',
'error': 'AKShare 未安装'}
def akshare_status(probe=False):
installed = importlib.util.find_spec('akshare') is not None
module = load_akshare() if probe and installed else _akshare_module
status = 'not_installed' if not installed else 'unobserved'
calendar = None
if probe and module is not None:
calendar = market_calendar_info()
status = 'ok' if calendar.get('confirmed') else 'degraded'
else:
with _source_lock:
observations = [dict(value) for key, value in _source_stats.items()
if key.startswith('akshare:')]
observation = max(observations, key=lambda row: row.get('last_at') or '') if observations else {}
if observation:
status = 'ok' if observation.get('ok') else 'degraded'
return {
'installed': installed,
'version': str(getattr(module, '__version__', '') or '') or None,
'status': status,
'calendar': calendar,
'role': '补充层:交易日历、宏观与跨市场公开数据;不提升为官方或实时主源',
'interfaces': {
'trade_calendar': bool(module and hasattr(module, 'tool_trade_date_hist_sina')),
'macro_calendar': bool(module and hasattr(module, 'macro_info_ws')),
'macro_corroboration': bool(module and hasattr(module, 'news_economic_baidu')),
'stock_news': bool(module and hasattr(module, 'stock_news_em')),
'research_snapshot': bool(module and build_akshare_research_snapshot),
},
'event_service': load_event_service_config() if 'load_event_service_config' in globals() else None,
'error': _akshare_error,
}
def _record_source(host, ok, latency_ms, error=''):
"""记录最近一次真实上游访问;状态页只展示观测事实,不主动探测或伪报在线。"""
with _source_lock:
prev = _source_stats.get(host, {})
failures = 0 if ok else int(prev.get('failures') or 0) + 1
_source_stats[host] = {
'ok': bool(ok), 'latency_ms': int(latency_ms),
'last_at': now_bj().isoformat(timespec='seconds'),
'last_ok': now_bj().isoformat(timespec='seconds') if ok else prev.get('last_ok'),
'failures': failures,
'error': '' if ok else str(error)[:160],
}
class UpstreamError(Exception):
pass
def fetch(url, timeout=9, encoding='utf-8', raw=False, retry=2, referer=None):
"""带限频与重试的上游 GET。返回文本(或 raw 时返回 bytes)。
上游偶发断开连接(RemoteDisconnected)时放慢重试,礼貌而坚韧。"""
host = urllib.parse.urlparse(url).netloc
for attempt in range(retry + 1):
with _host_lock:
wait = 0.2 - (time.monotonic() - _last_req.get(host, 0))
if wait > 0:
time.sleep(wait)
_last_req[host] = time.monotonic()
req = urllib.request.Request(url)
for k, v in UA_HEADERS.items():
req.add_header(k, v)
if referer:
req.add_header('Referer', referer)
started = time.monotonic()
try:
with urllib.request.urlopen(req, timeout=timeout) as resp:
body = resp.read()
_record_source(host, True, (time.monotonic() - started) * 1000)
if raw:
return body
return body.decode(encoding, 'replace')
except RemoteDisconnected:
time.sleep(1.2 * (attempt + 1)) # 上游断连:逐步放慢再试
except Exception as e:
if attempt >= retry:
_record_source(host, False, (time.monotonic() - started) * 1000, e)
raise UpstreamError('%s: %s' % (host, e))
time.sleep(0.5 * (attempt + 1))
raise UpstreamError(host + ': remote disconnected')
def json_loads(text):
"""容忍 JSONP 包装(如 var ajaxResult={...};)"""
t = text.strip()
if t.startswith('var '):
i = t.find('=')
t = t[i + 1:].strip()
t = t.rstrip(';').strip()
return json.loads(t)
def source_catalog():
"""返回来源分级与最近观测状态;未被请求的查验入口保持“未观测”。"""
with _source_lock:
stats = dict(_source_stats)
now_mono = time.monotonic()
items = []
for src in SOURCE_CATALOG:
observations = [stats[h] for h in src['hosts'] if h in stats]
last = max(observations, key=lambda x: x.get('last_at') or '') if observations else None
circuit = 0
if '_host_down' in globals():
circuit = max([max(0, int(_host_down.get(h, 0) - now_mono))
for h in src['hosts']] or [0])
environment = None
if src['id'] == 'tdx_local':
if not TDX_ENABLED:
status = 'disabled'
elif not tdx_local_api:
status = 'unavailable'
else:
environment = tdx_local_api.environment_status()
env_status = environment.get('status')
if env_status in ('unsupported', 'not_installed', 'not_running'):
status = env_status
elif last is not None:
status = 'ok' if last.get('ok') else 'unavailable'
else:
status = 'unobserved'
elif src['id'] == 'akshare':
environment = akshare_status(probe=False)
status = environment['status']
akshare_observations = [value for key, value in stats.items()
if key.startswith('akshare:')]
if akshare_observations:
last = max(akshare_observations,
key=lambda row: row.get('last_at') or '')
elif src['mode'] == 'reference':
status = 'reference'
elif circuit:
status = 'degraded'
elif last is None:
status = 'unobserved'
else:
status = 'ok' if last.get('ok') else 'degraded'
item = dict(src)
item.update({
'status': status,
'last_observed': last.get('last_at') if last else None,
'last_ok': last.get('last_ok') if last else None,
'latency_ms': last.get('latency_ms') if last else None,
'failures': last.get('failures') if last else 0,
'latest_ok': bool(last.get('ok')) if last else None,
'circuit_seconds': circuit,
})
if environment is not None:
item['environment'] = environment
items.append(item)
return {
'generated_at': now_bj().isoformat(timespec='seconds'),
'items': items,
'policy': '官方披露优先;通达信用于本地只读行情增强;AKShare 只作交易日历、宏观与跨市场补充;市场聚合用于备援与线索;未观测不等于可用。',
}
# ---------------------------------------------------------------- 缓存
_cache_lock = threading.Lock()
_cache = {}
def cached(key, ttl, fn):
"""TTL 内存缓存:key -> (expire_ts, value)"""
now = time.monotonic()
with _cache_lock:
hit = _cache.get(key)
if hit and hit[0] > now:
return hit[1]
value = fn()
with _cache_lock:
_cache[key] = (now + ttl, value)
return value
def cache_drop(prefix):
with _cache_lock:
for k in [k for k in _cache if k.startswith(prefix)]:
del _cache[k]
def normalize_akshare_research_preferences(value=None):
raw = value if isinstance(value, dict) else {}
selected = raw.get('enabledPacks')
if normalize_akshare_pack_ids:
selected = normalize_akshare_pack_ids(selected)
else:
selected = ['growth', 'prices', 'liquidity', 'rates']
return {
'modelVersion': 'akshare-research-preferences-v1',
'enabledPacks': selected,
'manualOnly': True,
'includedInEmotionScore': False,
'automaticTradingAction': False,
}
def load_akshare_research_preferences():
profile = load_profile().get('data') or {}
return normalize_akshare_research_preferences(profile.get('akshare_research_preferences'))
def save_akshare_research_preferences(value):
preferences = normalize_akshare_research_preferences(value)
saved = save_profile({'akshare_research_preferences': preferences})
cache_drop('akshare_research_snapshot_v2')
return {'profile': saved, 'preferences': preferences,
'catalog': akshare_pack_catalog() if akshare_pack_catalog else []}
def _akshare_interface_health(interface_names):
with _source_lock:
stats = dict(_source_stats)
rows = []
for name in interface_names:
observation = stats.get('akshare:' + str(name)) or {}
rows.append({
'interface': str(name),
'status': ('ok' if observation.get('ok') is True else
'failed' if observation.get('ok') is False else 'unobserved'),
'lastObserved': observation.get('last_at'), 'lastOk': observation.get('last_ok'),
'latencyMs': observation.get('latency_ms'),
'failures': int(observation.get('failures') or 0),
})
return rows
def akshare_research_snapshot(refresh=False, selected_packs=None):
"""Return a cached, on-demand snapshot for user-selected research packs."""
preferences = load_akshare_research_preferences()
selected = (normalize_akshare_pack_ids(selected_packs)
if normalize_akshare_pack_ids and selected_packs is not None
else preferences['enabledPacks'])
module = load_akshare()
version = str(getattr(module, '__version__', '') or '') if module else ''
if module is None or build_akshare_research_snapshot is None:
if unloaded_akshare_research_snapshot:
return unloaded_akshare_research_snapshot(False, version, selected)
return {'modelVersion': AKSHARE_RESEARCH_MODEL_VERSION, 'status': 'unavailable',
'modules': [], 'errors': [{'interface': 'akshare', 'error': _akshare_error or '不可用'}],
'includedInEmotionScore': False, 'automaticTradingAction': False}
key = 'akshare_research_snapshot_v2:' + ','.join(selected)
if refresh:
cache_drop(key)
else:
with _cache_lock:
hit = _cache.get(key)
if hit and hit[0] > time.monotonic():
return hit[1]
return unloaded_akshare_research_snapshot(True, version, selected)
def fetcher(name, **kwargs):
fn = getattr(module, name, None)
if not callable(fn):
raise RuntimeError('%s 接口不可用' % name)
started = time.monotonic()
try:
value = fn(**kwargs)
_record_source('akshare:' + name, True, (time.monotonic() - started) * 1000)
return value
except Exception as exc:
_record_source('akshare:' + name, False, (time.monotonic() - started) * 1000, exc)
raise
def loader():
snapshot = build_akshare_research_snapshot(fetcher, version, now_bj(), selected)
snapshot['interfaceHealth'] = _akshare_interface_health(snapshot.get('interfacesRequested') or [])
snapshot['preferences'] = preferences
return snapshot
return cached(key, 6 * 3600, loader)
# ---------------------------------------------------------------- 可选本地增强:通达信 TQ-Local(只读)
def tdx_status(probe=False, fresh=False):
"""返回通达信环境状态;probe=True 时按技能约束完成本地服务探测。"""
if not TDX_ENABLED:
return {
'supported': sys.platform == 'win32', 'installed': False,
'process_running': False, 'service_ready': False,
'status': 'disabled', 'read_only': True,
}
if not tdx_local_api:
return {
'supported': sys.platform == 'win32', 'installed': False,
'process_running': False, 'service_ready': False,
'status': 'unavailable', 'error': 'tdx_local adapter unavailable',
'read_only': True,
}
if not probe:
return tdx_local_api.environment_status()
if fresh:
cache_drop('tdx_status_probe')
def loader():
status = tdx_local_api.probe_status()
_record_source(TDX_HOST, status.get('service_ready', False),
status.get('latency_ms') or 0, status.get('error') or '')
return status
return cached('tdx_status_probe', 10, loader)
def _tdx_require_ready():
status = tdx_status(probe=True)
if status.get('service_ready'):
_clear_host_down(TDX_HOST)
return
if not _host_ok(TDX_HOST):
raise UpstreamError('通达信 TQ-Local 暂时熔断')
raise UpstreamError(status.get('error') or ('通达信状态:' + status.get('status', 'unavailable')))
def tdx_read_quote(code):
_tdx_require_ready()
started = time.monotonic()
try:
data = tdx_local_api.quote(code)
_record_source(TDX_HOST, True, data.get('latency_ms') or
(time.monotonic() - started) * 1000)
return data
except Exception as e:
_record_source(TDX_HOST, False, (time.monotonic() - started) * 1000, e)
_mark_host_down(TDX_HOST, 30)
raise UpstreamError(str(e))
def tdx_read_kline(code, n=320, klt=101, fqt=1, explicit_code=None):
_tdx_require_ready()
started = time.monotonic()
try:
data = tdx_local_api.kline(code, n, klt, fqt, explicit_code=explicit_code)
_record_source(TDX_HOST, True, data.get('latency_ms') or
(time.monotonic() - started) * 1000)
return data
except Exception as e:
_record_source(TDX_HOST, False, (time.monotonic() - started) * 1000, e)
_mark_host_down(TDX_HOST, 30)
raise UpstreamError(str(e))
def tdx_emotion_verification():
_tdx_require_ready()
started = time.monotonic()
try:
data = tdx_local_api.emotion_snapshot()
_record_source(TDX_HOST, True, data.get('latency_ms') or
(time.monotonic() - started) * 1000)
return data
except Exception as e:
# ErrorId 表示服务在线但当前客户端未提供该专业数据(常见于权限或本地数据未准备)。
# 这不应熔断仍然可用的实时行情和 K 线能力。
if 'ErrorId=' in str(e):
return {
'status': 'unavailable', 'source': 'tdx_local',
'source_name': '通达信 TQ-Local', 'read_only': True,
'fields': {}, 'error': str(e)[:240],
'reason': 'professional_market_data_unavailable',
}
_record_source(TDX_HOST, False, (time.monotonic() - started) * 1000, e)
_mark_host_down(TDX_HOST, 30)
raise UpstreamError(str(e))
# ---------------------------------------------------------------- 工具:代码/证券ID
def normalize_code(code):
"""把 sh600519 / SZ000001 / 600519.SH 等归一为纯数字代码"""
c = code.strip().lower()
c = re.sub(r'^(sh|sz|bj)\s*', '', c)
c = re.sub(r'\.(sh|sz|bj)$', '', c)
c = re.sub(r'[^0-9]', '', c)
return c
def secid_of(code):
"""数字代码 -> 东财 secid(1.沪 / 0.深)"""
if code[:1] in ('6', '5', '9'):
return '1.' + code
return '0.' + code
def tq_code_of(code):
"""数字代码 -> 腾讯代码(sh/sz 前缀)"""
return ('sh' if code[:1] in ('6', '5', '9') else 'sz') + code
# ---------------------------------------------------------------- 主数据源:东方财富
def em_indices(host='push2.eastmoney.com'):
url = ('https://%s/api/qt/ulist.np/get?fltt=2&invt=2'
'&secids=1.000001,0.399001,0.399006,1.000688,0.899050'
'&fields=f2,f3,f4,f6,f12,f14' % host)
j = json_loads(fetch(url))
out = []
for d in (j.get('data') or {}).get('diff') or []:
out.append({
'code': d.get('f12'), 'name': d.get('f14'),
'price': d.get('f2'), 'pct': d.get('f3'), 'chg': d.get('f4'),
'amount': d.get('f6') or 0,
})
return out
def em_indices_any():
"""指数实时行情(解析格式),push2 → push2delay 故障切换"""
last = None
for host in ('push2.eastmoney.com', 'push2delay.eastmoney.com'):
if not _host_ok(host):
continue
try:
url = ('https://%s/api/qt/ulist.np/get?fltt=2&invt=2'
'&secids=1.000001,0.399001,0.399006,1.000688,0.899050'
'&fields=f2,f3,f4,f6,f12,f14' % host)
j = json_loads(fetch(url))
out = []
for d in (j.get('data') or {}).get('diff') or []:
out.append({
'code': d.get('f12'), 'name': d.get('f14'),
'price': d.get('f2'), 'pct': d.get('f3'), 'chg': d.get('f4'),
'amount': d.get('f6') or 0,
})
return out
except Exception as e:
last = e
_mark_host_down(host)
raise last if last else UpstreamError('indices unavailable')
def em_ulist_any(secids, fields):
"""ulist 批量行情(原始 diff 行),push2 → push2delay 故障切换"""
last = None
for host in ('push2.eastmoney.com', 'push2delay.eastmoney.com'):
if not _host_ok(host):
continue
try:
url = ('https://%s/api/qt/ulist.np/get?fltt=2&invt=2&secids=%s&fields=%s'
% (host, secids, fields))
j = json_loads(fetch(url))
return (j.get('data') or {}).get('diff') or []
except Exception as e:
last = e
_mark_host_down(host)
raise last if last else UpstreamError('ulist unavailable')
def em_quote(secid, host='push2.eastmoney.com'):
"""个股实时行情。注意:本接口价格字段为 ×100(如 f43=134199 即 1341.99),
已与腾讯行情交叉验证;涨停池的 p 字段则是 ×1000,两者不同源需分别处理。"""
fields = 'f43,f44,f45,f46,f47,f48,f50,f57,f58,f60,f116,f117,f162,f167,f168,f169,f170'
url = ('https://%s/api/qt/stock/get?secid=%s&fields=%s' % (host, secid, fields))
j = json_loads(fetch(url))
d = j.get('data') or {}
if not d:
raise UpstreamError('no quote data')
return {
'code': d.get('f57'), 'name': d.get('f58'),
'price': (d.get('f43') or 0) / 100.0,
'high': (d.get('f44') or 0) / 100.0,
'low': (d.get('f45') or 0) / 100.0,
'open': (d.get('f46') or 0) / 100.0,
'volume': d.get('f47') or 0, # 手
'amount': d.get('f48') or 0, # 元
'vol_ratio': (d.get('f50') or 0) / 100.0, # 量比
'prev_close': (d.get('f60') or 0) / 100.0,
'mktcap': d.get('f116') or 0,
'float_mktcap': d.get('f117') or 0,
'pe': (d.get('f162') or 0) / 100.0,
'pb': (d.get('f167') or 0) / 100.0,
'turnover': (d.get('f168') or 0) / 100.0, # 换手率 %
'chg': (d.get('f169') or 0) / 100.0,
'pct': (d.get('f170') or 0) / 100.0,
}
def em_quote_any(secid):
"""多主机故障切换:push2 → push2delay(延迟镜像,限流时救急),带熔断"""
last = None
for host in ('push2.eastmoney.com', 'push2delay.eastmoney.com'):
if not _host_ok(host):
continue
try:
return em_quote(secid, host)
except Exception as e:
last = e
_mark_host_down(host)
raise last if last else UpstreamError('quote unavailable')
def em_kline(secid, klt=101, fqt=1, n=320, beg='20200101', host='push2his.eastmoney.com'):
url = ('https://%s/api/qt/stock/kline/get?secid=%s'
'&klt=%s&fqt=%s&beg=%s&end=20500101'
'&fields1=f1,f2,f3,f4,f5,f6'
'&fields2=f51,f52,f53,f54,f55,f56,f57,f58,f59,f60,f61'
% (host, secid, klt, fqt, beg))
j = json_loads(fetch(url))
d = j.get('data') or {}
rows = []
for line in (d.get('klines') or []):
p = line.split(',')
if len(p) < 11:
continue
rows.append({
'date': p[0], 'open': float(p[1]), 'close': float(p[2]),
'high': float(p[3]), 'low': float(p[4]),
'volume': float(p[5]), 'amount': float(p[6]),
'amp': float(p[7]), 'pct': float(p[8]),
'chg': float(p[9]), 'turn': float(p[10]),
})
if not rows:
raise UpstreamError('empty kline from %s' % host)
return {'name': d.get('name'), 'code': d.get('code'),
'pre': d.get('preKPrice'), 'rows': rows[-n:]}
def em_kline_any(secid, klt=101, fqt=1, n=320, beg='20200101'):
"""K线多主机故障切换:push2his → push2"""
last = None
for host in ('push2his.eastmoney.com', 'push2.eastmoney.com'):
if not _host_ok(host):
continue
try:
return em_kline(secid, klt, fqt, n, beg, host)
except Exception as e:
last = e
_mark_host_down(host)
raise last if last else UpstreamError('kline unavailable')
# 上游主机熔断:某主机连续失败后,短时间内不再尝试,快速切换备援
_host_down = {}
_host_down_lock = threading.Lock()
def _host_ok(host):
with _host_down_lock:
return time.monotonic() >= _host_down.get(host, 0)
def _mark_host_down(host, secs=180):
with _host_down_lock:
_host_down[host] = time.monotonic() + secs
log('circuit: %s marked down for %ds' % (host, secs))
def _clear_host_down(host):
with _host_down_lock:
_host_down.pop(host, None)
def em_pool(kind, date=None, size=250):
"""kind: ZT(涨停) / DT(跌停) / ZB(炸板);date: YYYYMMDD。
该接口必须携带 date 参数(缺省返回 rc=102);休市日自动回退到最近交易日。
注:pagesize 上限 250。"""
size = min(size, 250)
sort = 'fbt%3Aasc' if kind in ('ZT', 'ZB') else 'fund%3Aasc'
def _get(d):
url = ('https://push2ex.eastmoney.com/getTopic%sPool?ut=%s&dpt=wz.ztzt'
'&Pageindex=0&pagesize=%d&sort=%s&date=%s'
% (kind, EM_UT, size, sort, d))
return json_loads(fetch(url))
j = None
if date:
j = _get(date)
if not j or not (j.get('data') or {}).get('pool'):
# 回退:从今天往前找最近一个交易日(最多 10 天)
day = datetime.strptime(date or now_bj().strftime('%Y%m%d'), '%Y%m%d')
for i in range(1, 11):
d = (day - timedelta(days=i)).strftime('%Y%m%d')
try:
j2 = _get(d)
if (j2.get('data') or {}).get('pool'):
j = j2
break
except Exception:
continue
d = (j or {}).get('data') or {}
pool = []
for it in (d.get('pool') or []):
pool.append({
'code': it.get('c'), 'name': it.get('n'),
'price': (it.get('p') or 0) / 1000.0,
'pct': round(it.get('zdp') or 0, 2),
'amount': it.get('amount') or 0,
'float_mktcap': it.get('ltsz') or 0,
'turnover': round(it.get('hs') or 0, 2),
'lbc': it.get('lbc') or 0, # 连板数
'fbt': it.get('fbt'), # 首次封板时间 HHMMSS
'lbt': it.get('lbt'), # 最后封板时间
'fund': it.get('fund') or 0, # 封单资金
'zbc': it.get('zbc') or 0, # 炸板次数
'days': it.get('days') or 0, # 连续跌停天数(跌停池)
'hybk': it.get('hybk') or '', # 行业板块
'zttj': it.get('zttj') or {}, # 涨停统计 {days, ct}
})
return {'qdate': str(d.get('qdate') or ''), 'total': d.get('tc') or len(pool),
'pool': pool}
def em_breadth():
url = ('https://push2ex.eastmoney.com/getTopicZDFenBu?ut=%s&dpt=wz.ztzt' % EM_UT)
j = json_loads(fetch(url))
d = j.get('data') or {}
bins = {}
for item in (d.get('fenbu') or []):
bins.update(item)
up = sum(v for k, v in bins.items() if int(k) > 0)
down = sum(v for k, v in bins.items() if int(k) < 0)
flat = bins.get('0', 0)
limit_up = bins.get('11', 0)
limit_down = bins.get('-11', 0)