from __future__ import annotations
|
|
import argparse
|
import json
|
import sys
|
from datetime import datetime
|
from pathlib import Path
|
from typing import Any, Sequence
|
from zoneinfo import ZoneInfo
|
|
from .core import (
|
LedgerError,
|
active_securities,
|
apply_daily_prices,
|
export_latest,
|
import_valuations,
|
list_rows,
|
load_price_fixture,
|
show_ticker,
|
)
|
from .database import DatabaseConfigError, load_settings
|
|
|
PROJECT_ROOT = Path(__file__).resolve().parents[3]
|
DEFAULT_RESULTS_ROOT = PROJECT_ROOT / "ana-data" / "result" / "股票估值"
|
DEFAULT_LEDGER_DIR = DEFAULT_RESULTS_ROOT / "估值台账"
|
|
|
def _path(value: str) -> Path:
|
return Path(value).expanduser().resolve()
|
|
|
def _json(value: Any) -> None:
|
print(json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":"), default=str))
|
|
|
def _connection_arguments(parser: argparse.ArgumentParser) -> None:
|
parser.add_argument("--database", default="stock_valuation")
|
parser.add_argument("--config", type=_path, help="不进入仓库的本机 MySQL JSON 配置")
|
|
|
def build_parser() -> argparse.ArgumentParser:
|
parser = argparse.ArgumentParser(description="股票估值每日 MySQL 台账")
|
commands = parser.add_subparsers(dest="command", required=True)
|
|
init = commands.add_parser("init", help="创建 schema 并幂等导入正式估值版本")
|
init.add_argument("--results-root", type=_path, default=DEFAULT_RESULTS_ROOT)
|
init.add_argument("--project-root", type=_path, default=PROJECT_ROOT)
|
init.add_argument("--gaps", type=_path, default=DEFAULT_LEDGER_DIR / "import_gaps.csv")
|
_connection_arguments(init)
|
|
daily = commands.add_parser("daily", help="刷新完整交易日收盘价、判定和最新导出")
|
daily.add_argument("--as-of", default="latest")
|
daily.add_argument("--prices", type=_path, help="离线完整交易日行情 fixture")
|
daily.add_argument("--output-dir", type=_path, default=DEFAULT_LEDGER_DIR)
|
_connection_arguments(daily)
|
|
listing = commands.add_parser("list", help="列出最新或指定日期判定")
|
listing.add_argument("--date", default="latest")
|
listing.add_argument("--label")
|
listing.add_argument("--ticker")
|
listing.add_argument("--company")
|
_connection_arguments(listing)
|
|
show = commands.add_parser("show", help="显示单股全部版本、价格和判定历史")
|
show.add_argument("--ticker", required=True)
|
_connection_arguments(show)
|
|
export = commands.add_parser("export", help="生成 latest.md 和 latest.csv")
|
export.add_argument("--date", default="latest")
|
export.add_argument("--output-dir", type=_path, default=DEFAULT_LEDGER_DIR)
|
_connection_arguments(export)
|
return parser
|
|
|
def _as_of(value: str) -> str:
|
if value == "latest":
|
return datetime.now(ZoneInfo("Asia/Shanghai")).date().isoformat()
|
from datetime import date
|
|
return date.fromisoformat(value).isoformat()
|
|
|
def run(args: argparse.Namespace) -> tuple[int, dict[str, Any]]:
|
settings = load_settings(database=args.database, config_path=args.config)
|
if args.command == "init":
|
result = import_valuations(
|
settings,
|
args.results_root,
|
args.project_root,
|
args.gaps,
|
)
|
result.update({"database": settings.database, "results_root": str(args.results_root)})
|
return 0, result
|
if args.command == "daily":
|
as_of = _as_of(args.as_of)
|
securities = active_securities(settings)
|
if args.prices:
|
prices = load_price_fixture(args.prices)
|
present = {row.ticker for row in prices}
|
failures = [
|
{"ticker": item["ticker"], "code": "E_PRICE_MISSING", "message": "fixture 未提供行情"}
|
for item in securities
|
if item["ticker"] not in present
|
]
|
market_metrics = {"provider": "offline_fixture", "requested": len(securities), "succeeded": len(prices), "failed": len(failures)}
|
else:
|
from .market import fetch_trading_closes
|
|
prices, failures, market_metrics = fetch_trading_closes(settings, securities, as_of)
|
result = apply_daily_prices(settings, as_of, prices, failures)
|
result["market"] = market_metrics
|
if result["inserted_judgements"] + result["unchanged"] > 0:
|
result["export"] = export_latest(settings, args.output_dir, "latest")
|
else:
|
result["export"] = {
|
"status": "PRESERVED_NO_TRUSTED_JUDGEMENT",
|
"output_dir": str(args.output_dir),
|
}
|
return (4 if result["failure_count"] else 0), result
|
if args.command == "list":
|
rows = list_rows(
|
settings,
|
args.date,
|
label=args.label,
|
ticker=args.ticker,
|
company=args.company,
|
)
|
return 0, {"status": "OK", "count": len(rows), "rows": rows}
|
if args.command == "show":
|
return 0, {"status": "OK", **show_ticker(settings, args.ticker)}
|
if args.command == "export":
|
return 0, export_latest(settings, args.output_dir, args.date)
|
raise LedgerError("E_COMMAND", "未知命令")
|
|
|
def main(argv: Sequence[str] | None = None) -> int:
|
parser = build_parser()
|
try:
|
args = parser.parse_args(argv)
|
code, result = run(args)
|
_json(result)
|
return code
|
except (LedgerError, DatabaseConfigError) as exc:
|
code = exc.code if isinstance(exc, LedgerError) else "E_DATABASE_CONFIG"
|
_json({"status": "FAILED", "error_code": code, "error": str(exc)})
|
return 2
|
except (OSError, ValueError, json.JSONDecodeError) as exc:
|
_json({"status": "FAILED", "error_code": "E_INPUT", "error": str(exc)})
|
return 2
|
|
|
if __name__ == "__main__":
|
raise SystemExit(main(sys.argv[1:]))
|
