352
+ | def parse_datetime(value: str) -> datetime:
- """Parse an ISO 8601 datetime string to a timezone-aware datetime.
-
- Args:
- value: ISO 8601 datetime string (e.g., '2024-01-01' or
- '2024-01-01T12:00:00+00:00').
-
- Returns:
- Parsed datetime with UTC timezone if no timezone is specified.
-
- Raises:
- ValueError: If the string cannot be parsed.
- """
- try:
- dt = datetime.fromisoformat(value)
- except ValueError:
- msg = f"Invalid datetime format: '{value}'. Use ISO 8601 format."
- raise ValueError(msg) from None
- if dt.tzinfo is None:
- dt = dt.replace(tzinfo=UTC)
- return dt
+370
| def parse_datetime(value: str) -> datetime:
+ """Parse an ISO 8601 datetime string to a timezone-aware datetime.
+
+ Args:
+ value: ISO 8601 datetime string (e.g., '2024-01-01' or
+ '2024-01-01T12:00:00+00:00').
+
+ Returns:
+ Parsed datetime with UTC timezone if no timezone is specified.
+
+ Raises:
+ ValueError: If the string cannot be parsed.
+ """
+ try:
+ dt = datetime.fromisoformat(value)
+ except ValueError:
+ msg = f"Invalid datetime format: '{value}'. Use ISO 8601 format."
+ raise ValueError(msg) from None
+ if dt.tzinfo is None:
+ dt = dt.replace(tzinfo=UTC)
+ return dt
|
XAUUSDm, US500.cash, or EU
Source code in mt5cli/utils.py
- 399
+ | def parse_tick_flags(value: object) -> int:
- """Parse tick flags string or integer value.
-
- Args:
- value: Tick flag name (ALL, INFO, TRADE, COPY_TICKS_*) or integer value.
-
- Returns:
- Integer tick flag value compatible with MetaTrader 5 ``COPY_TICKS_*``.
-
- Raises:
- ValueError: If the flag is invalid.
- """
- try:
- return _parse_copy_ticks(value)
- except ValueError:
- display = value if isinstance(value, str) else repr(value)
- valid = ", ".join(_TICK_FLAG_NAMES)
- msg = (
- f"Invalid tick flags: '{display}'. "
- f"Use one of: {valid}, or a supported integer."
- )
- raise ValueError(msg) from None
+418
| def parse_tick_flags(value: object) -> int:
+ """Parse tick flags string or integer value.
+
+ Args:
+ value: Tick flag name (ALL, INFO, TRADE, COPY_TICKS_*) or integer value.
+
+ Returns:
+ Integer tick flag value compatible with MetaTrader 5 ``COPY_TICKS_*``.
+
+ Raises:
+ ValueError: If the flag is invalid.
+ """
+ try:
+ return _parse_copy_ticks(value)
+ except ValueError:
+ display = value if isinstance(value, str) else repr(value)
+ valid = ", ".join(_TICK_FLAG_NAMES)
+ msg = (
+ f"Invalid tick flags: '{display}'. "
+ f"Use one of: {valid}, or a supported integer."
+ )
+ raise ValueError(msg) from None
|
XAUUSDm, US500.cash, or EU
Source code in mt5cli/utils.py
- 375
+ | def parse_timeframe(value: object) -> int:
- """Parse a timeframe string or integer value.
-
- Args:
- value: Timeframe name (e.g., 'M1', 'H1', 'D1') or integer value.
-
- Returns:
- Integer timeframe value.
-
- Raises:
- ValueError: If the timeframe is invalid.
- """
- try:
- return _parse_timeframe(value)
- except ValueError:
- display = value if isinstance(value, str) else repr(value)
- valid = ", ".join(TIMEFRAME_NAMES)
- msg = (
- f"Invalid timeframe: '{display}'. "
- f"Use one of: {valid}, or a supported integer."
- )
- raise ValueError(msg) from None
+394
| def parse_timeframe(value: object) -> int:
+ """Parse a timeframe string or integer value.
+
+ Args:
+ value: Timeframe name (e.g., 'M1', 'H1', 'D1') or integer value.
+
+ Returns:
+ Integer timeframe value.
+
+ Raises:
+ ValueError: If the timeframe is invalid.
+ """
+ try:
+ return _parse_timeframe(value)
+ except ValueError:
+ display = value if isinstance(value, str) else repr(value)
+ valid = ", ".join(TIMEFRAME_NAMES)
+ msg = (
+ f"Invalid timeframe: '{display}'. "
+ f"Use one of: {valid}, or a supported integer."
+ )
+ raise ValueError(msg) from None
|
+ - Responsibility boundary
+
+
- Public API tiers
@@ -169,10 +173,39 @@
Public API Contract
- mt5cli is the generic MT5 data and execution infrastructure layer for downstream
-Python applications. The intended dependency direction is:
+ mt5cli is the canonical operational trading SDK and CLI/batch layer over pdmt5.
+The intended dependency direction is:
downstream app -> mt5cli -> pdmt5 -> MetaTrader 5
+ Responsibility boundary
+
+
+
+| Layer |
+Owns |
+
+
+
+
+| pdmt5 |
+MT5 core wrapper; DataFrame/dict conversion; canonical MT5 constants and parsers; direct low-level order primitives |
+
+
+| mt5cli |
+CLI/batch workflows; SQLite history collection; normalized datasets; closed-bar helpers; small downstream operational SDK; generic broker-facing margin/volume/order orchestration |
+
+
+| downstream |
+Strategy logic; signals; risk policy; backtesting; optimization; YAML/application semantics |
+
+
+
+ Downstream code should import raw pdmt5 types and constants (such as
+Mt5Config, Mt5TradingClient, Mt5RuntimeError, Mt5TradingError,
+TIMEFRAME_MAP, COPY_TICKS_MAP) directly from pdmt5 when needed.
+mt5cli does not serve as a pass-through compatibility namespace for pdmt5.
+ Note: the former mt5cli re-export TICK_FLAG_MAP corresponds to COPY_TICKS_MAP
+in pdmt5 — the name changed, it was not simply moved.
Downstream packages should import from the package root (from mt5cli import
...) and use the public tier sets in mt5cli.contract to distinguish API
stability. CLI commands mirror the same behavior but are not importable Python
@@ -448,7 +481,7 @@ after price movement and does not inspect trade_freeze_level. Live
sending requests. Failed, malformed, or unknown broker retcodes are fail-closed
and returned as status="failed" with normalized request / response details;
dry_run=True never calls ensure_symbol_selected() or order_send().
- Errors and MT5 type re-exports
+ Errors
@@ -465,10 +498,6 @@ and returned as status="failed" with normalized requestnormalize_mt5_exception, call_with_normalized_errors, is_recoverable_mt5_error
Error normalization and retry classification |
-
-Mt5Config, Mt5RuntimeError, Mt5TradingClient, Mt5TradingError |
-Re-exported pdmt5 types for adapter convenience |
-
Secondary public exports
@@ -548,7 +577,7 @@ the MT5 SDK helper.
|
| MT5 parsing maps |
-granularity_name, parse_tick_flags, parse_timeframe, TICK_FLAG_MAP, TIMEFRAME_MAP |
+granularity_name, parse_tick_flags, parse_timeframe |
| Trading data shapes |
diff --git a/api/storage/index.html b/api/storage/index.html
index d6ebe09..19134c2 100644
--- a/api/storage/index.html
+++ b/api/storage/index.html
@@ -1603,7 +1603,9 @@ a symbol such as EURUSD_M1 cannot collide with EURUSD
Source code in mt5cli/utils.py
- 216
+ | def detect_format(
- output_path: Path,
- explicit_format: str | None = None,
-) -> str:
- """Detect the output format from a file extension or explicit format string.
-
- Args:
- output_path: Path to the output file.
- explicit_format: Explicitly specified format, if any.
-
- Returns:
- The detected format string.
-
- Raises:
- ValueError: If the format cannot be determined.
- """
- if explicit_format is not None:
- return explicit_format
- suffix = output_path.suffix.lower()
- if suffix in _FORMAT_EXTENSIONS:
- return _FORMAT_EXTENSIONS[suffix]
- msg = (
- f"Cannot detect format from extension '{suffix}'."
- " Use --format to specify the output format."
- )
- raise ValueError(msg)
+239
| def detect_format(
+ output_path: Path,
+ explicit_format: str | None = None,
+) -> str:
+ """Detect the output format from a file extension or explicit format string.
+
+ Args:
+ output_path: Path to the output file.
+ explicit_format: Explicitly specified format, if any.
+
+ Returns:
+ The detected format string.
+
+ Raises:
+ ValueError: If the format cannot be determined.
+ """
+ if explicit_format is not None:
+ return explicit_format
+ suffix = output_path.suffix.lower()
+ if suffix in _FORMAT_EXTENSIONS:
+ return _FORMAT_EXTENSIONS[suffix]
+ msg = (
+ f"Cannot detect format from extension '{suffix}'."
+ " Use --format to specify the output format."
+ )
+ raise ValueError(msg)
|
Source code in mt5cli/utils.py
- 302
+ | def export_dataframe(
- df: pd.DataFrame,
- output_path: Path,
- output_format: str,
- table_name: str = "data",
-) -> None:
- """Export a pandas DataFrame to the specified file format.
-
- Args:
- df: DataFrame to export.
- output_path: Path to the output file.
- output_format: Output format (csv, json, parquet, or sqlite3).
- table_name: Table name for SQLite3 output.
-
- Raises:
- ImportError: If the parquet format is requested but pyarrow is not installed.
- ValueError: If the output format is not supported.
- """
- if output_format == "csv":
- df.to_csv(output_path, index=False)
- elif output_format == "json":
- df.to_json(
- output_path,
- orient="records",
- date_format="iso",
- indent=2,
- )
- elif output_format == "parquet":
- try:
- __import__("pyarrow")
- except ImportError as exc:
- msg = (
- "Parquet export requires the optional dependency pyarrow. "
- 'Install it with: pip install "mt5cli[parquet]"'
- )
- raise ImportError(msg) from exc
- df.to_parquet(output_path, index=False)
- elif output_format == "sqlite3":
- export_dataframe_to_sqlite(
- df,
- output_path,
- table_name,
- if_exists=IfExists.REPLACE,
- index=False,
- )
- else:
- msg = f"Unsupported output format: {output_format}"
- raise ValueError(msg)
+347
| def export_dataframe(
+ df: pd.DataFrame,
+ output_path: Path,
+ output_format: str,
+ table_name: str = "data",
+) -> None:
+ """Export a pandas DataFrame to the specified file format.
+
+ Args:
+ df: DataFrame to export.
+ output_path: Path to the output file.
+ output_format: Output format (csv, json, parquet, or sqlite3).
+ table_name: Table name for SQLite3 output.
+
+ Raises:
+ ImportError: If the parquet format is requested but pyarrow is not installed.
+ ValueError: If the output format is not supported.
+ """
+ if output_format == "csv":
+ df.to_csv(output_path, index=False)
+ elif output_format == "json":
+ df.to_json(
+ output_path,
+ orient="records",
+ date_format="iso",
+ indent=2,
+ )
+ elif output_format == "parquet":
+ try:
+ __import__("pyarrow")
+ except ImportError as exc:
+ msg = (
+ "Parquet export requires the optional dependency pyarrow. "
+ 'Install it with: pip install "mt5cli[parquet]"'
+ )
+ raise ImportError(msg) from exc
+ df.to_parquet(output_path, index=False)
+ elif output_format == "sqlite3":
+ export_dataframe_to_sqlite(
+ df,
+ output_path,
+ table_name,
+ if_exists=IfExists.REPLACE,
+ index=False,
+ )
+ else:
+ msg = f"Unsupported output format: {output_format}"
+ raise ValueError(msg)
|
Source code in mt5cli/utils.py
- 258
+ | def export_dataframe_to_sqlite(
- df: pd.DataFrame,
- output_path: Path,
- table_name: str = "data",
- *,
- if_exists: IfExists = IfExists.APPEND,
- index: bool = False,
- index_label: str | None = None,
- deduplicate_on: Sequence[str] | None = None,
-) -> None:
- """Write a DataFrame to SQLite with configurable append and deduplication.
-
- Args:
- df: DataFrame to export.
- output_path: SQLite database path.
- table_name: Target table name.
- if_exists: Conflict behavior when the table already exists.
- index: Whether to write the DataFrame index as a column.
- index_label: Column name for the index when ``index=True``.
- deduplicate_on: Optional key columns to deduplicate after writing,
- keeping the latest ``ROWID`` per key group. Deduplication scans the
- full table, so repeated appends cost O(table size); index the key
- columns when appending frequently.
- """
- with sqlite3.connect(output_path) as conn:
- df.to_sql( # type: ignore[reportUnknownMemberType]
- table_name,
- conn,
- if_exists=if_exists.value,
- index=index,
- index_label=index_label,
- )
- if deduplicate_on:
- from .history import drop_duplicates_in_table # noqa: PLC0415
-
- drop_duplicates_in_table(
- conn.cursor(),
- table_name,
- list(deduplicate_on),
- keep="last",
- )
- conn.commit()
+297
| def export_dataframe_to_sqlite(
+ df: pd.DataFrame,
+ output_path: Path,
+ table_name: str = "data",
+ *,
+ if_exists: IfExists = IfExists.APPEND,
+ index: bool = False,
+ index_label: str | None = None,
+ deduplicate_on: Sequence[str] | None = None,
+) -> None:
+ """Write a DataFrame to SQLite with configurable append and deduplication.
+
+ Args:
+ df: DataFrame to export.
+ output_path: SQLite database path.
+ table_name: Target table name.
+ if_exists: Conflict behavior when the table already exists.
+ index: Whether to write the DataFrame index as a column.
+ index_label: Column name for the index when ``index=True``.
+ deduplicate_on: Optional key columns to deduplicate after writing,
+ keeping the latest ``ROWID`` per key group. Deduplication scans the
+ full table, so repeated appends cost O(table size); index the key
+ columns when appending frequently.
+ """
+ with sqlite3.connect(output_path) as conn:
+ df.to_sql( # type: ignore[reportUnknownMemberType]
+ table_name,
+ conn,
+ if_exists=if_exists.value,
+ index=index,
+ index_label=index_label,
+ )
+ if deduplicate_on:
+ from .history import drop_duplicates_in_table # noqa: PLC0415
+
+ drop_duplicates_in_table(
+ conn.cursor(),
+ table_name,
+ list(deduplicate_on),
+ keep="last",
+ )
+ conn.commit()
|
- TICK_FLAG_MAP
-
-
-
- module-attribute
-
-
-
-TICK_FLAG_MAP: dict[str, int] = dict(COPY_TICKS_MAP)
-
-
-
-
-
-
-
-
-
-
TIMEFRAME_NAMES
@@ -919,7 +897,9 @@
Source code in mt5cli/utils.py
- 244
+ | def coerce_login(login: int | str | None) -> int | None:
- """Coerce a login value to int, treating empty strings as unset.
-
- Returns:
- Integer login, or None when unset or an empty string.
- """
- if login is None or isinstance(login, int):
- return login
- text = login.strip()
- if not text:
- return None
- return int(text)
+253
| def coerce_login(login: int | str | None) -> int | None:
+ """Coerce a login value to int, treating empty strings as unset.
+
+ Returns:
+ Integer login, or None when unset or an empty string.
+ """
+ if login is None or isinstance(login, int):
+ return login
+ text = login.strip()
+ if not text:
+ return None
+ return int(text)
|
Source code in mt5cli/utils.py
- 216
+ | def detect_format(
- output_path: Path,
- explicit_format: str | None = None,
-) -> str:
- """Detect the output format from a file extension or explicit format string.
-
- Args:
- output_path: Path to the output file.
- explicit_format: Explicitly specified format, if any.
-
- Returns:
- The detected format string.
-
- Raises:
- ValueError: If the format cannot be determined.
- """
- if explicit_format is not None:
- return explicit_format
- suffix = output_path.suffix.lower()
- if suffix in _FORMAT_EXTENSIONS:
- return _FORMAT_EXTENSIONS[suffix]
- msg = (
- f"Cannot detect format from extension '{suffix}'."
- " Use --format to specify the output format."
- )
- raise ValueError(msg)
+239
| def detect_format(
+ output_path: Path,
+ explicit_format: str | None = None,
+) -> str:
+ """Detect the output format from a file extension or explicit format string.
+
+ Args:
+ output_path: Path to the output file.
+ explicit_format: Explicitly specified format, if any.
+
+ Returns:
+ The detected format string.
+
+ Raises:
+ ValueError: If the format cannot be determined.
+ """
+ if explicit_format is not None:
+ return explicit_format
+ suffix = output_path.suffix.lower()
+ if suffix in _FORMAT_EXTENSIONS:
+ return _FORMAT_EXTENSIONS[suffix]
+ msg = (
+ f"Cannot detect format from extension '{suffix}'."
+ " Use --format to specify the output format."
+ )
+ raise ValueError(msg)
|
Source code in mt5cli/utils.py
- 302
+ | def export_dataframe(
- df: pd.DataFrame,
- output_path: Path,
- output_format: str,
- table_name: str = "data",
-) -> None:
- """Export a pandas DataFrame to the specified file format.
-
- Args:
- df: DataFrame to export.
- output_path: Path to the output file.
- output_format: Output format (csv, json, parquet, or sqlite3).
- table_name: Table name for SQLite3 output.
-
- Raises:
- ImportError: If the parquet format is requested but pyarrow is not installed.
- ValueError: If the output format is not supported.
- """
- if output_format == "csv":
- df.to_csv(output_path, index=False)
- elif output_format == "json":
- df.to_json(
- output_path,
- orient="records",
- date_format="iso",
- indent=2,
- )
- elif output_format == "parquet":
- try:
- __import__("pyarrow")
- except ImportError as exc:
- msg = (
- "Parquet export requires the optional dependency pyarrow. "
- 'Install it with: pip install "mt5cli[parquet]"'
- )
- raise ImportError(msg) from exc
- df.to_parquet(output_path, index=False)
- elif output_format == "sqlite3":
- export_dataframe_to_sqlite(
- df,
- output_path,
- table_name,
- if_exists=IfExists.REPLACE,
- index=False,
- )
- else:
- msg = f"Unsupported output format: {output_format}"
- raise ValueError(msg)
+347
| def export_dataframe(
+ df: pd.DataFrame,
+ output_path: Path,
+ output_format: str,
+ table_name: str = "data",
+) -> None:
+ """Export a pandas DataFrame to the specified file format.
+
+ Args:
+ df: DataFrame to export.
+ output_path: Path to the output file.
+ output_format: Output format (csv, json, parquet, or sqlite3).
+ table_name: Table name for SQLite3 output.
+
+ Raises:
+ ImportError: If the parquet format is requested but pyarrow is not installed.
+ ValueError: If the output format is not supported.
+ """
+ if output_format == "csv":
+ df.to_csv(output_path, index=False)
+ elif output_format == "json":
+ df.to_json(
+ output_path,
+ orient="records",
+ date_format="iso",
+ indent=2,
+ )
+ elif output_format == "parquet":
+ try:
+ __import__("pyarrow")
+ except ImportError as exc:
+ msg = (
+ "Parquet export requires the optional dependency pyarrow. "
+ 'Install it with: pip install "mt5cli[parquet]"'
+ )
+ raise ImportError(msg) from exc
+ df.to_parquet(output_path, index=False)
+ elif output_format == "sqlite3":
+ export_dataframe_to_sqlite(
+ df,
+ output_path,
+ table_name,
+ if_exists=IfExists.REPLACE,
+ index=False,
+ )
+ else:
+ msg = f"Unsupported output format: {output_format}"
+ raise ValueError(msg)
|
Source code in mt5cli/utils.py
- 258
+ | def export_dataframe_to_sqlite(
- df: pd.DataFrame,
- output_path: Path,
- table_name: str = "data",
- *,
- if_exists: IfExists = IfExists.APPEND,
- index: bool = False,
- index_label: str | None = None,
- deduplicate_on: Sequence[str] | None = None,
-) -> None:
- """Write a DataFrame to SQLite with configurable append and deduplication.
-
- Args:
- df: DataFrame to export.
- output_path: SQLite database path.
- table_name: Target table name.
- if_exists: Conflict behavior when the table already exists.
- index: Whether to write the DataFrame index as a column.
- index_label: Column name for the index when ``index=True``.
- deduplicate_on: Optional key columns to deduplicate after writing,
- keeping the latest ``ROWID`` per key group. Deduplication scans the
- full table, so repeated appends cost O(table size); index the key
- columns when appending frequently.
- """
- with sqlite3.connect(output_path) as conn:
- df.to_sql( # type: ignore[reportUnknownMemberType]
- table_name,
- conn,
- if_exists=if_exists.value,
- index=index,
- index_label=index_label,
- )
- if deduplicate_on:
- from .history import drop_duplicates_in_table # noqa: PLC0415
-
- drop_duplicates_in_table(
- conn.cursor(),
- table_name,
- list(deduplicate_on),
- keep="last",
- )
- conn.commit()
+297
| def export_dataframe_to_sqlite(
+ df: pd.DataFrame,
+ output_path: Path,
+ table_name: str = "data",
+ *,
+ if_exists: IfExists = IfExists.APPEND,
+ index: bool = False,
+ index_label: str | None = None,
+ deduplicate_on: Sequence[str] | None = None,
+) -> None:
+ """Write a DataFrame to SQLite with configurable append and deduplication.
+
+ Args:
+ df: DataFrame to export.
+ output_path: SQLite database path.
+ table_name: Target table name.
+ if_exists: Conflict behavior when the table already exists.
+ index: Whether to write the DataFrame index as a column.
+ index_label: Column name for the index when ``index=True``.
+ deduplicate_on: Optional key columns to deduplicate after writing,
+ keeping the latest ``ROWID`` per key group. Deduplication scans the
+ full table, so repeated appends cost O(table size); index the key
+ columns when appending frequently.
+ """
+ with sqlite3.connect(output_path) as conn:
+ df.to_sql( # type: ignore[reportUnknownMemberType]
+ table_name,
+ conn,
+ if_exists=if_exists.value,
+ index=index,
+ index_label=index_label,
+ )
+ if deduplicate_on:
+ from .history import drop_duplicates_in_table # noqa: PLC0415
+
+ drop_duplicates_in_table(
+ conn.cursor(),
+ table_name,
+ list(deduplicate_on),
+ keep="last",
+ )
+ conn.commit()
|
Source code in mt5cli/utils.py
- 352
+ | def parse_datetime(value: str) -> datetime:
- """Parse an ISO 8601 datetime string to a timezone-aware datetime.
-
- Args:
- value: ISO 8601 datetime string (e.g., '2024-01-01' or
- '2024-01-01T12:00:00+00:00').
-
- Returns:
- Parsed datetime with UTC timezone if no timezone is specified.
-
- Raises:
- ValueError: If the string cannot be parsed.
- """
- try:
- dt = datetime.fromisoformat(value)
- except ValueError:
- msg = f"Invalid datetime format: '{value}'. Use ISO 8601 format."
- raise ValueError(msg) from None
- if dt.tzinfo is None:
- dt = dt.replace(tzinfo=UTC)
- return dt
+370
| def parse_datetime(value: str) -> datetime:
+ """Parse an ISO 8601 datetime string to a timezone-aware datetime.
+
+ Args:
+ value: ISO 8601 datetime string (e.g., '2024-01-01' or
+ '2024-01-01T12:00:00+00:00').
+
+ Returns:
+ Parsed datetime with UTC timezone if no timezone is specified.
+
+ Raises:
+ ValueError: If the string cannot be parsed.
+ """
+ try:
+ dt = datetime.fromisoformat(value)
+ except ValueError:
+ msg = f"Invalid datetime format: '{value}'. Use ISO 8601 format."
+ raise ValueError(msg) from None
+ if dt.tzinfo is None:
+ dt = dt.replace(tzinfo=UTC)
+ return dt
|
Source code in mt5cli/utils.py
- 427
+ | def parse_request(value: str) -> dict[str, Any]:
- """Parse a JSON-formatted order request string or file reference.
-
- Args:
- value: JSON object string, or '@path' to read JSON from a file.
-
- Returns:
- Parsed request dictionary.
-
- Raises:
- ValueError: If the request file cannot be read or the value is not a
- JSON object.
- """
- if value.startswith("@"):
- path = Path(value[1:])
- try:
- text = path.read_text(encoding="utf-8")
- except (OSError, UnicodeDecodeError) as exc:
- msg = f"Failed to read JSON request file '{path}': {exc}"
- raise ValueError(msg) from exc
- else:
- text = value
- try:
- parsed: object = json.loads(text)
- except json.JSONDecodeError as exc:
- msg = f"Invalid JSON request: {exc}"
- raise ValueError(msg) from exc
- if not _is_request_dict(parsed):
- msg = "Order request must be a JSON object."
- raise ValueError(msg)
- return parsed
+455
| def parse_request(value: str) -> dict[str, Any]:
+ """Parse a JSON-formatted order request string or file reference.
+
+ Args:
+ value: JSON object string, or '@path' to read JSON from a file.
+
+ Returns:
+ Parsed request dictionary.
+
+ Raises:
+ ValueError: If the request file cannot be read or the value is not a
+ JSON object.
+ """
+ if value.startswith("@"):
+ path = Path(value[1:])
+ try:
+ text = path.read_text(encoding="utf-8")
+ except (OSError, UnicodeDecodeError) as exc:
+ msg = f"Failed to read JSON request file '{path}': {exc}"
+ raise ValueError(msg) from exc
+ else:
+ text = value
+ try:
+ parsed: object = json.loads(text)
+ except json.JSONDecodeError as exc:
+ msg = f"Invalid JSON request: {exc}"
+ raise ValueError(msg) from exc
+ if not _is_request_dict(parsed):
+ msg = "Order request must be a JSON object."
+ raise ValueError(msg)
+ return parsed
|
Source code in mt5cli/utils.py
- 399
+ | def parse_tick_flags(value: object) -> int:
- """Parse tick flags string or integer value.
-
- Args:
- value: Tick flag name (ALL, INFO, TRADE, COPY_TICKS_*) or integer value.
-
- Returns:
- Integer tick flag value compatible with MetaTrader 5 ``COPY_TICKS_*``.
-
- Raises:
- ValueError: If the flag is invalid.
- """
- try:
- return _parse_copy_ticks(value)
- except ValueError:
- display = value if isinstance(value, str) else repr(value)
- valid = ", ".join(_TICK_FLAG_NAMES)
- msg = (
- f"Invalid tick flags: '{display}'. "
- f"Use one of: {valid}, or a supported integer."
- )
- raise ValueError(msg) from None
+418
| def parse_tick_flags(value: object) -> int:
+ """Parse tick flags string or integer value.
+
+ Args:
+ value: Tick flag name (ALL, INFO, TRADE, COPY_TICKS_*) or integer value.
+
+ Returns:
+ Integer tick flag value compatible with MetaTrader 5 ``COPY_TICKS_*``.
+
+ Raises:
+ ValueError: If the flag is invalid.
+ """
+ try:
+ return _parse_copy_ticks(value)
+ except ValueError:
+ display = value if isinstance(value, str) else repr(value)
+ valid = ", ".join(_TICK_FLAG_NAMES)
+ msg = (
+ f"Invalid tick flags: '{display}'. "
+ f"Use one of: {valid}, or a supported integer."
+ )
+ raise ValueError(msg) from None
|
Source code in mt5cli/utils.py
- 375
+ | def parse_timeframe(value: object) -> int:
- """Parse a timeframe string or integer value.
-
- Args:
- value: Timeframe name (e.g., 'M1', 'H1', 'D1') or integer value.
-
- Returns:
- Integer timeframe value.
-
- Raises:
- ValueError: If the timeframe is invalid.
- """
- try:
- return _parse_timeframe(value)
- except ValueError:
- display = value if isinstance(value, str) else repr(value)
- valid = ", ".join(TIMEFRAME_NAMES)
- msg = (
- f"Invalid timeframe: '{display}'. "
- f"Use one of: {valid}, or a supported integer."
- )
- raise ValueError(msg) from None
+394
| def parse_timeframe(value: object) -> int:
+ """Parse a timeframe string or integer value.
+
+ Args:
+ value: Timeframe name (e.g., 'M1', 'H1', 'D1') or integer value.
+
+ Returns:
+ Integer timeframe value.
+
+ Raises:
+ ValueError: If the timeframe is invalid.
+ """
+ try:
+ return _parse_timeframe(value)
+ except ValueError:
+ display = value if isinstance(value, str) else repr(value)
+ valid = ", ".join(TIMEFRAME_NAMES)
+ msg = (
+ f"Invalid timeframe: '{display}'. "
+ f"Use one of: {valid}, or a supported integer."
+ )
+ raise ValueError(msg) from None
|
|
|
|
|
|
|
|
|
|
|
|
|
|